From d643ea416c128431939e08d7a54b38ad88b8ac71 Mon Sep 17 00:00:00 2001 From: Joshua Gilman Date: Mon, 24 Aug 2026 21:17:19 -0700 Subject: [PATCH] feat(worker): add bounded protocol framing --- internal/worker/decode.go | 263 +++++++++++ internal/worker/doc.go | 6 + internal/worker/frame.go | 844 +++++++++++++++++++++++++++++++++ internal/worker/frame_test.go | 848 ++++++++++++++++++++++++++++++++++ internal/worker/limits.go | 286 ++++++++++++ internal/worker/value.go | 195 ++++++++ internal/worker/value_test.go | 170 +++++++ 7 files changed, 2612 insertions(+) create mode 100644 internal/worker/decode.go create mode 100644 internal/worker/doc.go create mode 100644 internal/worker/frame.go create mode 100644 internal/worker/frame_test.go create mode 100644 internal/worker/limits.go create mode 100644 internal/worker/value.go create mode 100644 internal/worker/value_test.go diff --git a/internal/worker/decode.go b/internal/worker/decode.go new file mode 100644 index 0000000..33c2df9 --- /dev/null +++ b/internal/worker/decode.go @@ -0,0 +1,263 @@ +package worker + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "strconv" + "strings" +) + +// rawTypedFrame is the discriminator-only preview used before concrete decode. +type rawTypedFrame struct { + // Type is the discriminator used to select a concrete frame. + Type string `json:"type"` +} + +// rawVersionFrame is the versioned handshake object for probe and probe_ack. +type rawVersionFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // Version is the optional protocol version token. + Version *int `json:"version"` +} + +// rawExecFrame is the raw exec object used to distinguish absent from null fields. +type rawExecFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // Version is the protocol version. + Version *int `json:"version"` + + // Source is the Starlark program token. + Source *string `json:"source"` + + // Limits is the raw child-limit object. + Limits json.RawMessage `json:"limits"` + + // Manifest is the raw capability array. + Manifest json.RawMessage `json:"manifest"` +} + +// rawNativeCallFrame is the raw native_call object used to require arguments. +type rawNativeCallFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // CapabilityID is the capability identity token. + CapabilityID *string `json:"capability_id"` + + // Arguments is the raw argument object. + Arguments json.RawMessage `json:"arguments"` +} + +// rawResultFrame is the raw result object used to require a present result token. +type rawResultFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // Result is the raw normalized value token. + Result json.RawMessage `json:"result"` +} + +// rawFinalErrorFrame is the raw final_error object used to require a code. +type rawFinalErrorFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // Code is the child-owned error class. + Code *finalErrorCode `json:"code"` +} + +// decodeType reads only the type discriminator without rejecting unknown fields. +func decodeType(payload []byte) (string, error) { + if len(payload) == 0 { + return "", errMalformedJSON + } + dec := json.NewDecoder(bytes.NewReader(payload)) + dec.UseNumber() + var preview rawTypedFrame + if err := dec.Decode(&preview); err != nil { + return "", classifyJSONErr(err) + } + if err := requireEOF(dec); err != nil { + return "", err + } + return preview.Type, nil +} + +// decodeVersioned decodes probe or probe_ack and requires protocol version 1. +func decodeVersioned(payload []byte, wantType string) (probeFrame, error) { + var raw rawVersionFrame + if err := decodeStrict(payload, &raw); err != nil { + return probeFrame{}, err + } + if raw.Type != wantType { + return probeFrame{}, errMalformedJSON + } + if err := requireProtocolVersion(raw.Version); err != nil { + return probeFrame{}, err + } + return probeFrame{Type: wantType, Version: protocolVersion}, nil +} + +// decodeExec decodes one versioned execution request. +func decodeExec(payload []byte) (execFrame, error) { + var raw rawExecFrame + if err := decodeStrict(payload, &raw); err != nil { + return execFrame{}, err + } + if raw.Type != frameTypeExec { + return execFrame{}, errMalformedJSON + } + if err := requireProtocolVersion(raw.Version); err != nil { + return execFrame{}, err + } + if raw.Source == nil || isJSONNull(raw.Limits) || isJSONNull(raw.Manifest) { + return execFrame{}, errInvalidValue + } + var limits childLimits + if err := decodeStrict(raw.Limits, &limits); err != nil { + return execFrame{}, err + } + if err := validateChildLimits(limits); err != nil { + return execFrame{}, err + } + var manifest []manifestEntry + if err := decodeStrict(raw.Manifest, &manifest); err != nil { + return execFrame{}, err + } + if err := validateManifest(manifest); err != nil { + return execFrame{}, err + } + return execFrame{ + Type: frameTypeExec, + Version: protocolVersion, + Source: *raw.Source, + Limits: limits, + Manifest: manifest, + }, nil +} + +// decodeNativeCall decodes one native invocation with a required argument object. +func decodeNativeCall(payload []byte) (nativeCallFrame, error) { + var raw rawNativeCallFrame + if err := decodeStrict(payload, &raw); err != nil { + return nativeCallFrame{}, err + } + if raw.Type != frameTypeNativeCall { + return nativeCallFrame{}, errMalformedJSON + } + if raw.CapabilityID == nil || *raw.CapabilityID == "" || isJSONNull(raw.Arguments) { + return nativeCallFrame{}, errInvalidValue + } + value, err := decodeNormalizedValue(raw.Arguments) + if err != nil { + return nativeCallFrame{}, err + } + arguments, ok := value.(map[string]any) + if !ok { + return nativeCallFrame{}, errInvalidValue + } + return nativeCallFrame{ + Type: frameTypeNativeCall, + CapabilityID: *raw.CapabilityID, + Arguments: arguments, + }, nil +} + +// decodeResult decodes a required result token, including explicit null. +func decodeResult(payload []byte, wantType string) (any, error) { + var raw rawResultFrame + if err := decodeStrict(payload, &raw); err != nil { + return nil, err + } + if raw.Type != wantType { + return nil, errMalformedJSON + } + if raw.Result == nil { + return nil, errInvalidValue + } + return decodeNormalizedValue(raw.Result) +} + +// decodeFinalError decodes one known child-owned terminal code. +func decodeFinalError(payload []byte) (finalErrorFrame, error) { + var raw rawFinalErrorFrame + if err := decodeStrict(payload, &raw); err != nil { + return finalErrorFrame{}, err + } + if raw.Type != frameTypeFinalError { + return finalErrorFrame{}, errMalformedJSON + } + if raw.Code == nil || !knownFinalError(*raw.Code) { + return finalErrorFrame{}, errInvalidValue + } + return finalErrorFrame{Type: frameTypeFinalError, Code: *raw.Code}, nil +} + +// requireProtocolVersion rejects a missing or mismatched protocol version. +func requireProtocolVersion(version *int) error { + const mismatch = "%w: protocol version mismatch: child reported %s, parent requires %d" + if version == nil { + return fmt.Errorf(mismatch, errInvalidValue, "", protocolVersion) + } + if *version != protocolVersion { + return fmt.Errorf(mismatch, errInvalidValue, strconv.Itoa(*version), protocolVersion) + } + return nil +} + +// decodeStrict decodes exactly one JSON value and rejects unknown fields. +func decodeStrict(payload []byte, dest any) error { + if len(payload) == 0 { + return errMalformedJSON + } + dec := json.NewDecoder(bytes.NewReader(payload)) + dec.UseNumber() + dec.DisallowUnknownFields() + if err := dec.Decode(dest); err != nil { + return classifyJSONErr(err) + } + return requireEOF(dec) +} + +// requireEOF rejects a trailing JSON value after the one legal object. +func requireEOF(dec *json.Decoder) error { + if _, err := dec.Token(); err != io.EOF { + if err == nil { + return errTrailingJSON + } + return errTrailingJSON + } + return nil +} + +// classifyJSONErr maps encoding/json failures onto protocol sentinels. +func classifyJSONErr(err error) error { + if err == nil { + return nil + } + message := err.Error() + if strings.Contains(message, "unknown field") { + return errUnknownField + } + var syntax *json.SyntaxError + if errors.As(err, &syntax) { + return errMalformedJSON + } + var unmarshalType *json.UnmarshalTypeError + if errors.As(err, &unmarshalType) { + return errMalformedJSON + } + return errMalformedJSON +} + +// isJSONNull reports whether raw is absent or an explicit JSON null token. +func isJSONNull(raw json.RawMessage) bool { + return len(raw) == 0 || bytes.Equal(raw, []byte("null")) +} diff --git a/internal/worker/doc.go b/internal/worker/doc.go new file mode 100644 index 0000000..6711d9e --- /dev/null +++ b/internal/worker/doc.go @@ -0,0 +1,6 @@ +// Package worker implements the private same-executable parent/child transport. +// +// Protocol types, frame codecs, and connection state machines are not public +// API. They exist only between a host process and a worker re-exec of the same +// binary. +package worker diff --git a/internal/worker/frame.go b/internal/worker/frame.go new file mode 100644 index 0000000..607ea21 --- /dev/null +++ b/internal/worker/frame.go @@ -0,0 +1,844 @@ +package worker + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "errors" + "io" + "unicode/utf8" +) + +const protocolVersion = 1 + +const ( + frameTypeProbe = "probe" + frameTypeProbeAck = "probe_ack" + frameTypeExec = "exec" + frameTypeNativeCall = "native_call" + frameTypeNativeResult = "native_result" + frameTypeNativeAbort = "native_abort" + frameTypeFinal = "final" + frameTypeFinalError = "final_error" +) + +// finalErrorCode is one child-owned terminal failure class. +type finalErrorCode string + +const ( + finalErrorInvalidProgram finalErrorCode = "invalid_program" + finalErrorInvalidArguments finalErrorCode = "invalid_arguments" + finalErrorResourceLimit finalErrorCode = "resource_limit" + finalErrorInternal finalErrorCode = "internal" +) + +// knownFinalError reports whether code is one child-owned terminal class. +func knownFinalError(code finalErrorCode) bool { + switch code { + case finalErrorInvalidProgram, finalErrorInvalidArguments, finalErrorResourceLimit, finalErrorInternal: + return true + default: + return false + } +} + +const ( + // probeCap is the exact compact {"type":"probe_ack","version":1} size. + probeCap uint32 = 32 + + framePrefixSize = 4 +) + +var ( + errEmptyPayload = errors.New("empty payload") + errTruncatedFrame = errors.New("truncated frame") + errFrameTooLarge = errors.New("frame too large") + errInvalidUTF8 = errors.New("invalid utf-8") + errMalformedJSON = errors.New("malformed json") + errTrailingJSON = errors.New("trailing json") + errUnknownType = errors.New("unknown frame type") + errUnknownField = errors.New("unknown field") + errInvalidNumber = errors.New("invalid number") + errIllegalState = errors.New("illegal protocol state") + errInvalidValue = errors.New("invalid value") + errInvalidLimits = errors.New("invalid limits") + errInvalidManifest = errors.New("invalid manifest") +) + +const ( + nativeCallPrefix = `{"type":"native_call","capability_id":` + nativeCallMid = `,"arguments":` + nativeCallSuffix = `}` + finalPrefix = `{"type":"final","result":` + finalSuffix = `}` + finalErrorLongest = `{"type":"final_error","code":"invalid_arguments"}` + nativeResultPrefix = `{"type":"native_result","result":` + nativeResultSuffix = `}` + nativeAbortPayload = `{"type":"native_abort"}` + emptyJSONString = `""` +) + +// probeFrame is the parent's first and only probe request. +type probeFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // Version is the private protocol version. + Version int `json:"version"` +} + +// probeAckFrame is the child's successful probe reply. +type probeAckFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // Version is the private protocol version. + Version int `json:"version"` +} + +// execFrame is the parent's first execution request. +type execFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // Version is the private protocol version. + Version int `json:"version"` + + // Source is the Starlark program. + Source string `json:"source"` + + // Limits are the child-visible execution budgets. + Limits childLimits `json:"limits"` + + // Manifest lists enabled capabilities for this child. + Manifest []manifestEntry `json:"manifest"` +} + +// nativeCallFrame is one child-originated capability invocation. +type nativeCallFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // CapabilityID is the enabled capability identity. + CapabilityID string `json:"capability_id"` + + // Arguments is the fresh canonical argument map. + Arguments map[string]any `json:"arguments"` +} + +// nativeResultFrame is the parent's successful native response. +type nativeResultFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // Result is one normalized process-neutral value. + Result any `json:"result"` +} + +// nativeAbortFrame is the parent's terminal native failure. +type nativeAbortFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` +} + +// finalFrame is the child's successful terminal result. +type finalFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // Result is one normalized process-neutral value. + Result any `json:"result"` +} + +// finalErrorFrame is the child's terminal classified failure. +type finalErrorFrame struct { + // Type is the frame discriminator. + Type string `json:"type"` + + // Code is one child-owned final error class. + Code finalErrorCode `json:"code"` +} + +type connKind uint8 + +const ( + connKindProbe connKind = iota + connKindExec +) + +type connState uint8 + +const ( + stateInit connState = iota + stateAwaitPeer + stateReady + stateAwaitNative + stateDone +) + +// parentConn is the parent side of one probe or execution exchange. +type parentConn struct { + // w receives parent-originated frames. + w io.Writer + + // r supplies child-originated frames. + r io.Reader + + // writeCap bounds parent-originated payloads after the handshake. + writeCap uint32 + + // readCap bounds untrusted child-originated payloads. + readCap uint32 + + // kind selects probe or execution rules. + kind connKind + + // state is the current protocol phase. + state connState + + // nativeCalls is the number of accepted native_call frames. + nativeCalls uint64 + + // maxNativeCalls caps accepted native_call frames. + maxNativeCalls uint64 + + // limits are the child budgets from the written exec frame. + limits childLimits + + // hasLimits reports whether exec limits have been stored. + hasLimits bool +} + +// childConn is the child side of one probe or execution exchange. +type childConn struct { + // r supplies parent-originated frames. + r io.Reader + + // w receives child-originated frames. + w io.Writer + + // execReadCap bounds the initial trusted exec payload. + execReadCap uint32 + + // parentReadCap bounds later parent-originated payloads. + parentReadCap uint32 + + // writeCap bounds child-originated payloads. + writeCap uint32 + + // kind selects probe or execution rules. + kind connKind + + // state is the current protocol phase. + state connState + + // nativeCalls is the number of written native_call frames. + nativeCalls uint64 + + // limits are the child budgets from the received exec frame. + limits childLimits + + // hasLimits reports whether exec limits have been stored. + hasLimits bool +} + +// writeFrame writes one length-prefixed payload after checking the bound. +func writeFrame(w io.Writer, payload []byte, maxPayload uint32) error { + if w == nil { + return errIllegalState + } + if len(payload) == 0 { + return errEmptyPayload + } + if err := rejectOverCap(len(payload), maxPayload); err != nil { + return err + } + length, err := uint32FromInt(len(payload)) + if err != nil { + return err + } + var prefix [framePrefixSize]byte + binary.BigEndian.PutUint32(prefix[:], length) + if err := writeFull(w, prefix[:]); err != nil { + return err + } + return writeFull(w, payload) +} + +// readFrame reads one length-prefixed payload after checking the bound. +func readFrame(r io.Reader, maxPayload uint32) ([]byte, error) { + if r == nil { + return nil, errIllegalState + } + var prefix [framePrefixSize]byte + if err := readFull(r, prefix[:]); err != nil { + return nil, classifyReadErr(err) + } + length := binary.BigEndian.Uint32(prefix[:]) + if length == 0 { + return nil, errEmptyPayload + } + if length > maxPayload { + return nil, errFrameTooLarge + } + payload := make([]byte, length) + if err := readFull(r, payload); err != nil { + return nil, classifyReadErr(err) + } + return payload, nil +} + +// encodeProbe encodes the compact protocol-1 probe request. +func encodeProbe() ([]byte, error) { + return marshalFrame(probeFrame{Type: frameTypeProbe, Version: protocolVersion}) +} + +// encodeProbeAck encodes the compact protocol-1 probe acknowledgement. +func encodeProbeAck() ([]byte, error) { + return marshalFrame(probeAckFrame{Type: frameTypeProbeAck, Version: protocolVersion}) +} + +// encodeExec encodes one execution request after validating limits and manifest. +func encodeExec(frame execFrame) ([]byte, error) { + if err := validateChildLimits(frame.Limits); err != nil { + return nil, err + } + if err := validateManifest(frame.Manifest); err != nil { + return nil, err + } + if frame.Manifest == nil { + frame.Manifest = []manifestEntry{} + } + frame.Type = frameTypeExec + frame.Version = protocolVersion + return marshalFrame(frame) +} + +// encodeNativeCall encodes one child-originated native invocation. +func encodeNativeCall(capabilityID string, arguments map[string]any) ([]byte, error) { + if capabilityID == "" { + return nil, errInvalidValue + } + if arguments == nil { + return nil, errInvalidValue + } + encodedArgs, err := encodeNormalizedValue(arguments) + if err != nil { + return nil, err + } + encodedID, err := json.Marshal(capabilityID) + if err != nil { + return nil, errInvalidValue + } + var buf bytes.Buffer + buf.Grow(len(nativeCallPrefix) + len(encodedID) + len(nativeCallMid) + len(encodedArgs) + len(nativeCallSuffix)) + buf.WriteString(nativeCallPrefix) + buf.Write(encodedID) + buf.WriteString(nativeCallMid) + buf.Write(encodedArgs) + buf.WriteString(nativeCallSuffix) + return buf.Bytes(), nil +} + +// encodeNativeResult encodes one parent-originated native success. +func encodeNativeResult(result any) ([]byte, error) { + encoded, err := encodeNormalizedValue(result) + if err != nil { + return nil, err + } + var buf bytes.Buffer + buf.Grow(len(nativeResultPrefix) + len(encoded) + len(nativeResultSuffix)) + buf.WriteString(nativeResultPrefix) + buf.Write(encoded) + buf.WriteString(nativeResultSuffix) + return buf.Bytes(), nil +} + +// encodeNativeAbort encodes the payload-free parent abort frame. +func encodeNativeAbort() ([]byte, error) { + return []byte(nativeAbortPayload), nil +} + +// encodeFinal encodes one successful terminal child result. +func encodeFinal(result any) ([]byte, error) { + encoded, err := encodeNormalizedValue(result) + if err != nil { + return nil, err + } + var buf bytes.Buffer + buf.Grow(len(finalPrefix) + len(encoded) + len(finalSuffix)) + buf.WriteString(finalPrefix) + buf.Write(encoded) + buf.WriteString(finalSuffix) + return buf.Bytes(), nil +} + +// encodeFinalError encodes one child-owned terminal failure. +func encodeFinalError(code finalErrorCode) ([]byte, error) { + if !knownFinalError(code) { + return nil, errInvalidValue + } + return marshalFrame(finalErrorFrame{Type: frameTypeFinalError, Code: code}) +} + +// decodePayload decodes exactly one UTF-8 JSON frame object. +func decodePayload(payload []byte) (any, error) { + if !utf8.Valid(payload) { + return nil, errInvalidUTF8 + } + previewType, err := decodeType(payload) + if err != nil { + return nil, err + } + switch previewType { + case frameTypeProbe: + frame, err := decodeVersioned(payload, frameTypeProbe) + if err != nil { + return nil, err + } + return frame, nil + case frameTypeProbeAck: + frame, err := decodeVersioned(payload, frameTypeProbeAck) + if err != nil { + return nil, err + } + return probeAckFrame(frame), nil + case frameTypeExec: + return decodeExec(payload) + case frameTypeNativeCall: + return decodeNativeCall(payload) + case frameTypeNativeResult: + result, err := decodeResult(payload, frameTypeNativeResult) + if err != nil { + return nil, err + } + return nativeResultFrame{Type: frameTypeNativeResult, Result: result}, nil + case frameTypeNativeAbort: + if err := decodeStrict(payload, &nativeAbortFrame{}); err != nil { + return nil, err + } + return nativeAbortFrame{Type: frameTypeNativeAbort}, nil + case frameTypeFinal: + result, err := decodeResult(payload, frameTypeFinal) + if err != nil { + return nil, err + } + return finalFrame{Type: frameTypeFinal, Result: result}, nil + case frameTypeFinalError: + return decodeFinalError(payload) + case "": + return nil, errMalformedJSON + default: + return nil, errUnknownType + } +} + +// newParentProbeConn constructs the parent side of a probe exchange. +func newParentProbeConn(w io.Writer, r io.Reader) *parentConn { + return &parentConn{ + w: w, + r: r, + writeCap: probeCap, + readCap: probeCap, + kind: connKindProbe, + state: stateInit, + } +} + +// newChildProbeConn constructs the child side of a probe exchange. +func newChildProbeConn(r io.Reader, w io.Writer) *childConn { + return &childConn{ + r: r, + w: w, + execReadCap: probeCap, + parentReadCap: probeCap, + writeCap: probeCap, + kind: connKindProbe, + state: stateInit, + } +} + +// newParentExecConn constructs the parent side of an execution exchange. +func newParentExecConn(w io.Writer, r io.Reader, writeCap uint32, readCap uint32) *parentConn { + return &parentConn{ + w: w, + r: r, + writeCap: writeCap, + readCap: readCap, + kind: connKindExec, + state: stateInit, + } +} + +// newChildExecConn constructs the child side of an execution exchange. +func newChildExecConn(r io.Reader, w io.Writer, execReadCap uint32, parentReadCap uint32, writeCap uint32) *childConn { + return &childConn{ + r: r, + w: w, + execReadCap: execReadCap, + parentReadCap: parentReadCap, + writeCap: writeCap, + kind: connKindExec, + state: stateInit, + } +} + +// writeProbe writes the first and only parent probe request. +func (c *parentConn) writeProbe() error { + if c == nil || c.kind != connKindProbe || c.state != stateInit { + return errIllegalState + } + payload, err := encodeProbe() + if err != nil { + return err + } + if err := writeFrame(c.w, payload, c.writeCap); err != nil { + return err + } + c.state = stateAwaitPeer + return nil +} + +// writeExec writes the first parent execution request. +func (c *parentConn) writeExec(frame execFrame) error { + if c == nil || c.kind != connKindExec || c.state != stateInit { + return errIllegalState + } + payload, err := encodeExec(frame) + if err != nil { + return err + } + maxPayload, err := execPayloadCap(frame.Limits.MaxSourceBytes, frame.Manifest) + if err != nil { + return err + } + if err := writeFrame(c.w, payload, maxPayload); err != nil { + return err + } + c.limits = frame.Limits + c.hasLimits = true + c.maxNativeCalls = frame.Limits.MaxNativeCalls + c.state = stateReady + return nil +} + +// writeNativeResult writes the response to the outstanding native call. +func (c *parentConn) writeNativeResult(result any) error { + if c == nil || c.kind != connKindExec || c.state != stateAwaitNative || !c.hasLimits { + return errIllegalState + } + if err := validateBoundedValue(result, c.limits); err != nil { + return err + } + payload, err := encodeNativeResult(result) + if err != nil { + return err + } + if err := writeFrame(c.w, payload, c.writeCap); err != nil { + return err + } + c.state = stateReady + return nil +} + +// writeNativeAbort writes the terminal parent-owned native failure. +func (c *parentConn) writeNativeAbort() error { + if c == nil || c.kind != connKindExec || c.state != stateAwaitNative { + return errIllegalState + } + payload, err := encodeNativeAbort() + if err != nil { + return err + } + if err := writeFrame(c.w, payload, c.writeCap); err != nil { + return err + } + c.state = stateDone + return nil +} + +// read consumes the next legal child frame. +func (c *parentConn) read() (any, error) { + if c == nil || c.state != stateAwaitPeer && c.state != stateReady { + return nil, errIllegalState + } + payload, err := readFrame(c.r, c.readCap) + if err != nil { + return nil, err + } + frame, err := decodePayload(payload) + if err != nil { + return nil, err + } + switch typed := frame.(type) { + case probeAckFrame: + return c.acceptProbeAck(typed) + case nativeCallFrame: + return c.acceptNativeCall(typed) + case finalFrame: + return c.acceptFinal(typed) + case finalErrorFrame: + return c.acceptFinalError(typed) + default: + return nil, errIllegalState + } +} + +// acceptProbeAck records a successful probe acknowledgement. +func (c *parentConn) acceptProbeAck(frame probeAckFrame) (any, error) { + if c.kind != connKindProbe || c.state != stateAwaitPeer || frame.Version != protocolVersion { + return nil, errIllegalState + } + c.state = stateDone + return frame, nil +} + +// acceptNativeCall records one in-budget child native invocation. +func (c *parentConn) acceptNativeCall(frame nativeCallFrame) (any, error) { + if c.kind != connKindExec || c.state != stateReady || !c.hasLimits { + return nil, errIllegalState + } + if c.nativeCalls >= c.maxNativeCalls { + return nil, errIllegalState + } + if err := validateBoundedValue(frame.Arguments, c.limits); err != nil { + return nil, err + } + c.nativeCalls++ + c.state = stateAwaitNative + return frame, nil +} + +// acceptFinal records a successful child terminal result. +func (c *parentConn) acceptFinal(frame finalFrame) (any, error) { + if c.kind != connKindExec || c.state != stateReady || !c.hasLimits { + return nil, errIllegalState + } + if err := validateBoundedValue(frame.Result, c.limits); err != nil { + return nil, err + } + c.state = stateDone + return frame, nil +} + +// acceptFinalError records a legal child-owned terminal failure. +func (c *parentConn) acceptFinalError(frame finalErrorFrame) (any, error) { + if c.kind != connKindExec || c.state != stateReady { + return nil, errIllegalState + } + c.state = stateDone + return frame, nil +} + +// writeProbeAck writes the child's successful probe reply. +func (c *childConn) writeProbeAck() error { + if c == nil || c.kind != connKindProbe || c.state != stateReady { + return errIllegalState + } + payload, err := encodeProbeAck() + if err != nil { + return err + } + if err := writeFrame(c.w, payload, c.writeCap); err != nil { + return err + } + c.state = stateDone + return nil +} + +// writeNativeCall writes one native invocation and waits for a parent reply. +func (c *childConn) writeNativeCall(capabilityID string, arguments map[string]any) error { + if c == nil || c.kind != connKindExec || c.state != stateReady || !c.hasLimits { + return errIllegalState + } + if c.nativeCalls >= c.limits.MaxNativeCalls { + return errIllegalState + } + if err := validateBoundedValue(arguments, c.limits); err != nil { + return err + } + payload, err := encodeNativeCall(capabilityID, arguments) + if err != nil { + return err + } + if err := writeFrame(c.w, payload, c.writeCap); err != nil { + return err + } + c.nativeCalls++ + c.state = stateAwaitNative + return nil +} + +// writeFinal writes the successful terminal child result. +func (c *childConn) writeFinal(result any) error { + if c == nil || c.kind != connKindExec || c.state != stateReady || !c.hasLimits { + return errIllegalState + } + if err := validateBoundedValue(result, c.limits); err != nil { + return err + } + payload, err := encodeFinal(result) + if err != nil { + return err + } + if err := writeFrame(c.w, payload, c.writeCap); err != nil { + return err + } + c.state = stateDone + return nil +} + +// writeFinalError writes one child-owned terminal failure. +func (c *childConn) writeFinalError(code finalErrorCode) error { + if c == nil || c.kind != connKindExec || c.state != stateReady { + return errIllegalState + } + payload, err := encodeFinalError(code) + if err != nil { + return err + } + if err := writeFrame(c.w, payload, c.writeCap); err != nil { + return err + } + c.state = stateDone + return nil +} + +// read consumes the next legal parent frame. +func (c *childConn) read() (any, error) { + if c == nil { + return nil, errIllegalState + } + maxPayload, err := c.readLimit() + if err != nil { + return nil, err + } + payload, err := readFrame(c.r, maxPayload) + if err != nil { + return nil, err + } + frame, err := decodePayload(payload) + if err != nil { + return nil, err + } + switch typed := frame.(type) { + case probeFrame: + return c.acceptProbe(typed) + case execFrame: + return c.acceptExec(typed) + case nativeResultFrame: + return c.acceptNativeResult(typed) + case nativeAbortFrame: + return c.acceptNativeAbort(typed) + default: + return nil, errIllegalState + } +} + +// readLimit returns the state-specific parent payload bound. +func (c *childConn) readLimit() (uint32, error) { + switch { + case c.kind == connKindProbe && c.state == stateInit: + return c.execReadCap, nil + case c.kind == connKindExec && c.state == stateInit: + return c.execReadCap, nil + case c.kind == connKindExec && c.state == stateAwaitNative: + return c.parentReadCap, nil + default: + return 0, errIllegalState + } +} + +// acceptProbe records the first probe request. +func (c *childConn) acceptProbe(frame probeFrame) (any, error) { + if c.kind != connKindProbe || c.state != stateInit || frame.Version != protocolVersion { + return nil, errIllegalState + } + c.state = stateReady + return frame, nil +} + +// acceptExec records the first execution request and stores child limits. +func (c *childConn) acceptExec(frame execFrame) (any, error) { + if c.kind != connKindExec || c.state != stateInit { + return nil, errIllegalState + } + c.limits = frame.Limits + c.hasLimits = true + c.state = stateReady + return frame, nil +} + +// acceptNativeResult records the response to the outstanding native call. +func (c *childConn) acceptNativeResult(frame nativeResultFrame) (any, error) { + if c.kind != connKindExec || c.state != stateAwaitNative || !c.hasLimits { + return nil, errIllegalState + } + if err := validateBoundedValue(frame.Result, c.limits); err != nil { + return nil, err + } + c.state = stateReady + return frame, nil +} + +// acceptNativeAbort records the terminal parent-owned native failure. +func (c *childConn) acceptNativeAbort(frame nativeAbortFrame) (any, error) { + if c.kind != connKindExec || c.state != stateAwaitNative { + return nil, errIllegalState + } + c.state = stateDone + return frame, nil +} + +// classifyReadErr maps stream failures onto truncated-frame protocol errors. +func classifyReadErr(err error) error { + if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { + return errTruncatedFrame + } + return err +} + +// writeFull writes every byte or fails. +func writeFull(w io.Writer, payload []byte) error { + written, err := w.Write(payload) + if err != nil { + return err + } + if written != len(payload) { + return io.ErrShortWrite + } + return nil +} + +// readFull reads every requested byte or fails. +func readFull(r io.Reader, payload []byte) error { + _, err := io.ReadFull(r, payload) + return err +} + +// rejectOverCap rejects a payload that cannot fit the uint32 prefix or bound. +func rejectOverCap(length int, maxPayload uint32) error { + if length < 0 { + return errFrameTooLarge + } + converted, err := uint32FromInt(length) + if err != nil { + return err + } + if converted > maxPayload { + return errFrameTooLarge + } + return nil +} + +// marshalFrame encodes one concrete frame without a trailing newline. +func marshalFrame(value any) ([]byte, error) { + payload, err := json.Marshal(value) + if err != nil { + return nil, errMalformedJSON + } + if len(payload) == 0 { + return nil, errEmptyPayload + } + return payload, nil +} diff --git a/internal/worker/frame_test.go b/internal/worker/frame_test.go new file mode 100644 index 0000000..a85137f --- /dev/null +++ b/internal/worker/frame_test.go @@ -0,0 +1,848 @@ +package worker + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "errors" + "io" + "math" + "strings" + "sync" + "testing" + "unicode/utf8" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/meigma/codemode/internal/binding" +) + +const ( + testWriteCap uint32 = 4096 + testReadCap uint32 = 4096 +) + +// TestFrameWriteAndReadRoundTrip proves a length-prefixed payload survives an exact cap. +func TestFrameWriteAndReadRoundTrip(t *testing.T) { + payload, err := encodeProbeAck() + require.NoError(t, err) + require.Len(t, payload, int(probeCap), "probe_ack must be the compact 32-byte handshake") + + var buf bytes.Buffer + require.NoError(t, writeFrame(&buf, payload, probeCap)) + require.Equal(t, 4+int(probeCap), buf.Len()) + assert.Equal(t, uint32(len(payload)), binary.BigEndian.Uint32(buf.Bytes()[:4])) + + got, err := readFrame(&buf, probeCap) + require.NoError(t, err) + assert.Equal(t, payload, got) +} + +// TestFrameRejectsInvalidLengths proves caps are checked before a payload is accepted. +func TestFrameRejectsInvalidLengths(t *testing.T) { + valid, err := encodeProbe() + require.NoError(t, err) + + tests := []struct { + // name identifies the framing failure. + name string + + // raw is the complete byte stream presented to the reader. + raw []byte + + // cap is the state-specific payload cap. + cap uint32 + + // target is the expected protocol error. + target error + }{ + {name: "zero length", raw: prefixed(t, nil), cap: probeCap, target: errEmptyPayload}, + {name: "truncated prefix", raw: []byte{0x00, 0x00}, cap: probeCap, target: errTruncatedFrame}, + { + name: "truncated payload", + raw: append(u32(uint32(len(valid))), valid[:len(valid)/2]...), + cap: probeCap, + target: errTruncatedFrame, + }, + { + name: "over-cap length", + raw: prefixed(t, bytes.Repeat([]byte("a"), int(probeCap)+1)), + cap: probeCap, + target: errFrameTooLarge, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := readFrame(bytes.NewReader(tt.raw), tt.cap) + + require.Error(t, err) + require.ErrorIs(t, err, tt.target) + }) + } +} + +// TestFrameWriteRejectsOverCapPayloads proves writers measure the encoded payload before prefixing. +func TestFrameWriteRejectsOverCapPayloads(t *testing.T) { + var buf bytes.Buffer + err := writeFrame(&buf, bytes.Repeat([]byte("x"), int(probeCap)+1), probeCap) + + require.Error(t, err) + require.ErrorIs(t, err, errFrameTooLarge) + assert.Zero(t, buf.Len(), "an over-cap write must not emit a prefix") +} + +// TestFrameRejectsInvalidPayloads proves UTF-8 is checked before a strict single-object JSON decode. +func TestFrameRejectsInvalidPayloads(t *testing.T) { + tests := []struct { + // name identifies the decode failure. + name string + + // payload is the unframed JSON object. + payload string + + // target is the expected protocol error. + target error + }{ + {name: "invalid UTF-8", payload: "{\xff\"type\":\"probe\",\"version\":1}", target: errInvalidUTF8}, + {name: "malformed JSON", payload: `{"type":"probe","version":`, target: errMalformedJSON}, + {name: "trailing document", payload: `{"type":"probe","version":1}{}`, target: errTrailingJSON}, + {name: "unknown type", payload: `{"type":"nope","version":1}`, target: errUnknownType}, + {name: "unknown field", payload: `{"type":"probe","version":1,"extra":true}`, target: errUnknownField}, + {name: "wrong version", payload: `{"type":"probe","version":2}`, target: errInvalidValue}, + {name: "missing version", payload: `{"type":"probe"}`, target: errInvalidValue}, + {name: "null version", payload: `{"type":"probe","version":null}`, target: errInvalidValue}, + {name: "version on later frame", payload: `{"type":"native_abort","version":1}`, target: errUnknownField}, + { + name: "unknown final code", + payload: `{"type":"final_error","code":"permission_denied"}`, + target: errInvalidValue, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.True(t, tt.name != "invalid UTF-8" || !utf8.ValidString(tt.payload)) + _, err := decodeFramed(t, tt.payload) + + require.Error(t, err) + require.ErrorIs(t, err, tt.target) + if tt.name == "wrong version" { + assert.Contains(t, err.Error(), "2") + } + }) + } +} + +// TestFrameRequiredFieldsDistinguishNull proves explicit JSON null is not treated as an omitted field. +func TestFrameRequiredFieldsDistinguishNull(t *testing.T) { + t.Run("explicit null result is legal", func(t *testing.T) { + decoded, err := decodeFramed(t, `{"type":"final","result":null}`) + require.NoError(t, err) + frame, ok := decoded.(finalFrame) + require.True(t, ok) + assert.Equal(t, frameTypeFinal, frame.Type) + assert.Nil(t, frame.Result) + }) + + t.Run("absent result is rejected", func(t *testing.T) { + _, err := decodeFramed(t, `{"type":"final"}`) + require.Error(t, err) + require.ErrorIs(t, err, errInvalidValue) + }) + + t.Run("explicit null native result is legal", func(t *testing.T) { + decoded, err := decodeFramed(t, `{"type":"native_result","result":null}`) + require.NoError(t, err) + frame, ok := decoded.(nativeResultFrame) + require.True(t, ok) + assert.Nil(t, frame.Result) + }) + + t.Run("absent native result is rejected", func(t *testing.T) { + _, err := decodeFramed(t, `{"type":"native_result"}`) + require.Error(t, err) + require.ErrorIs(t, err, errInvalidValue) + }) + + t.Run("absent native arguments is rejected", func(t *testing.T) { + _, err := decodeFramed(t, `{"type":"native_call","capability_id":"cap.lookup"}`) + require.Error(t, err) + require.ErrorIs(t, err, errInvalidValue) + }) + + t.Run("absent final error code is rejected", func(t *testing.T) { + _, err := decodeFramed(t, `{"type":"final_error"}`) + require.Error(t, err) + require.ErrorIs(t, err, errInvalidValue) + }) +} + +// TestFrameEncodersProduceCompactDiscriminators proves every protocol type is privately named. +func TestFrameEncodersProduceCompactDiscriminators(t *testing.T) { + exec := validExecFrame() + nativeArgs := map[string]any{"org": "meigma"} + + tests := []struct { + // name identifies the encoder. + name string + + // encode produces one compact payload. + encode func() ([]byte, error) + + // typ is the frozen discriminator. + typ string + }{ + {name: "probe", encode: encodeProbe, typ: frameTypeProbe}, + {name: "probe_ack", encode: encodeProbeAck, typ: frameTypeProbeAck}, + {name: "exec", encode: func() ([]byte, error) { return encodeExec(exec) }, typ: frameTypeExec}, + { + name: "native_call", + encode: func() ([]byte, error) { return encodeNativeCall("cap.lookup", nativeArgs) }, + typ: frameTypeNativeCall, + }, + { + name: "native_result", + encode: func() ([]byte, error) { return encodeNativeResult(nil) }, + typ: frameTypeNativeResult, + }, + {name: "native_abort", encode: encodeNativeAbort, typ: frameTypeNativeAbort}, + {name: "final", encode: func() ([]byte, error) { return encodeFinal("ok") }, typ: frameTypeFinal}, + { + name: "final_error", + encode: func() ([]byte, error) { return encodeFinalError(finalErrorInvalidProgram) }, + typ: frameTypeFinalError, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + payload, err := tt.encode() + require.NoError(t, err) + assert.True(t, json.Valid(payload)) + assert.Contains(t, string(payload), `"type":"`+tt.typ+`"`) + + decoded, err := decodePayload(payload) + require.NoError(t, err) + assertFrameType(t, decoded, tt.typ) + }) + } +} + +func TestProtocolProbeSucceeds(t *testing.T) { + parent, child, closePipes := newProbePair(t) + + var wg sync.WaitGroup + wg.Go(func() { + defer closePipes() + frame, err := child.read() + if err != nil { + t.Errorf("child.read() error = %v", err) + return + } + assertFrameType(t, frame, frameTypeProbe) + if err := child.writeProbeAck(); err != nil { + t.Errorf("child.writeProbeAck() error = %v", err) + } + }) + + require.NoError(t, parent.writeProbe()) + ack, err := parent.read() + require.NoError(t, err) + assertFrameType(t, ack, frameTypeProbeAck) + wg.Wait() + + _, err = parent.read() + require.Error(t, err) + require.ErrorIs(t, err, errIllegalState) +} + +// TestProtocolAcceptsChildResourceLimit proves final_error(resource_limit) is a legal child terminal report. +func TestProtocolAcceptsChildResourceLimit(t *testing.T) { + parent, child, closePipes := newExecPair(t, validExecFrame()) + + var wg sync.WaitGroup + wg.Go(func() { + defer closePipes() + _, err := child.read() + if err != nil { + t.Errorf("child.read() error = %v", err) + return + } + if err := child.writeFinalError(finalErrorResourceLimit); err != nil { + t.Errorf("child.writeFinalError() error = %v", err) + } + }) + + require.NoError(t, parent.writeExec(validExecFrame())) + frame, err := parent.read() + require.NoError(t, err) + got, ok := frame.(finalErrorFrame) + require.True(t, ok) + assert.Equal(t, finalErrorResourceLimit, got.Code) + wg.Wait() +} + +func TestProtocolExecWithNativeCalls(t *testing.T) { + parent, child, closePipes := newExecPair(t, validExecFrame()) + + var wg sync.WaitGroup + wg.Go(func() { + defer closePipes() + frame, err := child.read() + if err != nil { + t.Errorf("child.read() error = %v", err) + return + } + assertFrameType(t, frame, frameTypeExec) + + for _, result := range []any{"one", "two"} { + err := child.writeNativeCall("cap.lookup", map[string]any{"org": "meigma"}) + if err != nil { + t.Errorf("child.writeNativeCall() error = %v", err) + return + } + reply, err := child.read() + if err != nil { + t.Errorf("child.read() error = %v", err) + return + } + got, ok := reply.(nativeResultFrame) + if !ok { + t.Errorf("child.read() type = %T, want nativeResultFrame", reply) + return + } + assert.Equal(t, result, got.Result) + } + if err := child.writeFinal(map[string]any{"ok": true}); err != nil { + t.Errorf("child.writeFinal() error = %v", err) + } + }) + + require.NoError(t, parent.writeExec(validExecFrame())) + for _, result := range []any{"one", "two"} { + call, err := parent.read() + require.NoError(t, err) + got, ok := call.(nativeCallFrame) + require.True(t, ok) + assert.Equal(t, "cap.lookup", got.CapabilityID) + assert.Equal(t, map[string]any{"org": "meigma"}, got.Arguments) + require.NoError(t, parent.writeNativeResult(result)) + } + final, err := parent.read() + require.NoError(t, err) + got, ok := final.(finalFrame) + require.True(t, ok) + assert.Equal(t, map[string]any{"ok": true}, got.Result) + wg.Wait() +} + +// TestProtocolTerminalFinalErrorAndAbort proves both terminal tails close the exchange. +func TestProtocolTerminalFinalErrorAndAbort(t *testing.T) { + t.Run("child final_error", func(t *testing.T) { + parent, child, closePipes := newExecPair(t, validExecFrame()) + + var wg sync.WaitGroup + wg.Go(func() { + defer closePipes() + _, err := child.read() + if err != nil { + t.Errorf("child.read() error = %v", err) + return + } + if err := child.writeFinalError(finalErrorInvalidArguments); err != nil { + t.Errorf("child.writeFinalError() error = %v", err) + } + }) + + require.NoError(t, parent.writeExec(validExecFrame())) + frame, err := parent.read() + require.NoError(t, err) + got, ok := frame.(finalErrorFrame) + require.True(t, ok) + assert.Equal(t, finalErrorInvalidArguments, got.Code) + wg.Wait() + }) + + t.Run("parent abort", func(t *testing.T) { + parent, child, closePipes := newExecPair(t, validExecFrame()) + + var wg sync.WaitGroup + wg.Go(func() { + defer closePipes() + _, err := child.read() + if err != nil { + t.Errorf("child.read() error = %v", err) + return + } + err = child.writeNativeCall("cap.lookup", map[string]any{"org": "meigma"}) + if err != nil { + t.Errorf("child.writeNativeCall() error = %v", err) + return + } + reply, err := child.read() + if err != nil { + t.Errorf("child.read() error = %v", err) + return + } + assertFrameType(t, reply, frameTypeNativeAbort) + }) + + require.NoError(t, parent.writeExec(validExecFrame())) + _, err := parent.read() + require.NoError(t, err) + require.NoError(t, parent.writeNativeAbort()) + wg.Wait() + }) +} + +// TestProtocolRejectsIllegalStateTransitions proves the state machine has no call IDs or extra frames. +func TestProtocolRejectsIllegalStateTransitions(t *testing.T) { + tests := []struct { + // name identifies the illegal transition. + name string + + // run drives one side into the forbidden state. + run func(t *testing.T) + }{ + { + name: "native result without outstanding call", + run: func(t *testing.T) { + parent, child := newBufferedExecPair(t, validExecFrame()) + require.NoError(t, parent.writeExec(validExecFrame())) + _, err := child.read() + require.NoError(t, err) + err = parent.writeNativeResult("late") + require.Error(t, err) + require.ErrorIs(t, err, errIllegalState) + }, + }, + { + name: "second outstanding native call", + run: func(t *testing.T) { + parent, child := newBufferedExecPair(t, validExecFrame()) + require.NoError(t, parent.writeExec(validExecFrame())) + _, err := child.read() + require.NoError(t, err) + require.NoError(t, child.writeNativeCall("cap.lookup", map[string]any{"org": "meigma"})) + err = child.writeNativeCall("cap.other", map[string]any{"org": "other"}) + require.Error(t, err) + require.ErrorIs(t, err, errIllegalState) + }, + }, + { + name: "frame after final", + run: func(t *testing.T) { + parent, child := newBufferedExecPair(t, validExecFrame()) + require.NoError(t, parent.writeExec(validExecFrame())) + _, err := child.read() + require.NoError(t, err) + require.NoError(t, child.writeFinal(nil)) + _, err = parent.read() + require.NoError(t, err) + err = child.writeFinalError(finalErrorInternal) + require.Error(t, err) + require.ErrorIs(t, err, errIllegalState) + }, + }, + { + name: "frame after abort", + run: func(t *testing.T) { + parent, child := newBufferedExecPair(t, validExecFrame()) + require.NoError(t, parent.writeExec(validExecFrame())) + _, err := child.read() + require.NoError(t, err) + require.NoError(t, child.writeNativeCall("cap.lookup", map[string]any{"org": "meigma"})) + _, err = parent.read() + require.NoError(t, err) + require.NoError(t, parent.writeNativeAbort()) + _, err = child.read() + require.NoError(t, err) + err = child.writeFinalError(finalErrorInternal) + require.Error(t, err) + require.ErrorIs(t, err, errIllegalState) + }, + }, + { + name: "probe frame on exec connection", + run: func(t *testing.T) { + parent, _ := newBufferedExecPair(t, validExecFrame()) + err := parent.writeProbe() + require.Error(t, err) + require.ErrorIs(t, err, errIllegalState) + }, + }, + { + name: "native call beyond MaxNativeCalls", + run: func(t *testing.T) { + exec := validExecFrame() + exec.Limits.MaxNativeCalls = 1 + parent, child := newBufferedExecPair(t, exec) + require.NoError(t, parent.writeExec(exec)) + _, err := child.read() + require.NoError(t, err) + require.NoError(t, child.writeNativeCall("cap.lookup", map[string]any{"org": "meigma"})) + _, err = parent.read() + require.NoError(t, err) + require.NoError(t, parent.writeNativeResult("ok")) + _, err = child.read() + require.NoError(t, err) + err = child.writeNativeCall("cap.lookup", map[string]any{"org": "meigma"}) + require.Error(t, err) + require.ErrorIs(t, err, errIllegalState) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, tt.run) + } +} + +// TestProtocolChildUsesExecCapThenParentCap proves the first child read is independently bounded. +func TestProtocolChildUsesExecCapThenParentCap(t *testing.T) { + exec := validExecFrame() + execPayload, err := encodeExec(exec) + require.NoError(t, err) + execCap, err := execPayloadCap(exec.Limits.MaxSourceBytes, exec.Manifest) + require.NoError(t, err) + parentCap, err := parentPayloadCap(exec.Limits.MaxValueBytes) + require.NoError(t, err) + require.Greater(t, execCap, parentCap) + + var toChild bytes.Buffer + child := newChildExecConn(&toChild, io.Discard, execCap, parentCap, testWriteCap) + require.NoError(t, writeFrame(&toChild, execPayload, execCap)) + frame, err := child.read() + require.NoError(t, err) + assertFrameType(t, frame, frameTypeExec) + require.NoError(t, child.writeNativeCall("cap.lookup", map[string]any{"org": "meigma"})) + + oversized := []byte(`{"type":"native_abort","pad":"` + strings.Repeat("a", int(parentCap)) + `"}`) + require.NoError(t, writeFrame(&toChild, oversized, execCap)) + _, err = child.read() + require.Error(t, err) + require.ErrorIs(t, err, errFrameTooLarge) +} + +// TestFrameLimitsValidateChildLimits proves zero never means unlimited. +func TestFrameLimitsValidateChildLimits(t *testing.T) { + valid := validExecFrame().Limits + + tests := []struct { + // name identifies the rejected limit. + name string + + // mutate changes one positive field. + mutate func(*childLimits) + }{ + {name: "zero source", mutate: func(limits *childLimits) { limits.MaxSourceBytes = 0 }}, + {name: "negative source", mutate: func(limits *childLimits) { limits.MaxSourceBytes = -1 }}, + {name: "zero steps", mutate: func(limits *childLimits) { limits.MaxExecutionSteps = 0 }}, + {name: "zero native calls", mutate: func(limits *childLimits) { limits.MaxNativeCalls = 0 }}, + {name: "zero depth", mutate: func(limits *childLimits) { limits.MaxValueDepth = 0 }}, + {name: "zero value bytes", mutate: func(limits *childLimits) { limits.MaxValueBytes = 0 }}, + } + + require.NoError(t, validateChildLimits(valid)) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + limits := valid + tt.mutate(&limits) + err := validateChildLimits(limits) + require.Error(t, err) + require.ErrorIs(t, err, errInvalidLimits) + }) + } +} + +// TestFrameLimitsValidateManifest proves the child cannot invent IDs, names, or field shapes. +func TestFrameLimitsValidateManifest(t *testing.T) { + valid := validManifest() + + tests := []struct { + // name identifies the rejected manifest. + name string + + // manifest is the candidate child descriptor. + manifest []manifestEntry + }{ + {name: "blank ID", manifest: []manifestEntry{{ID: "", Name: "records.lookup", Input: valid[0].Input}}}, + { + name: "whitespace ID", + manifest: []manifestEntry{{ID: " cap.lookup", Name: "records.lookup", Input: valid[0].Input}}, + }, + {name: "duplicate ID", manifest: []manifestEntry{ + {ID: "cap.lookup", Name: "records.lookup", Input: valid[0].Input}, + {ID: "cap.lookup", Name: "records.other", Input: valid[0].Input}, + }}, + {name: "undotted name", manifest: []manifestEntry{{ID: "cap.lookup", Name: "records", Input: valid[0].Input}}}, + { + name: "invalid name", + manifest: []manifestEntry{{ID: "cap.lookup", Name: "records.not-valid", Input: valid[0].Input}}, + }, + { + name: "reserved name segment", + manifest: []manifestEntry{{ID: "cap.lookup", Name: "records.import", Input: valid[0].Input}}, + }, + {name: "duplicate name", manifest: []manifestEntry{ + {ID: "cap.one", Name: "records.lookup", Input: valid[0].Input}, + {ID: "cap.two", Name: "records.lookup", Input: valid[0].Input}, + }}, + {name: "namespace collision", manifest: []manifestEntry{ + {ID: "cap.lookup", Name: "records.lookup", Input: valid[0].Input}, + {ID: "cap.detail", Name: "records.lookup.detail", Input: valid[0].Input}, + }}, + {name: "unsupported input shape", manifest: []manifestEntry{{ + ID: "cap.lookup", + Name: "records.lookup", + Input: []binding.FieldShape{{ + Name: "org", + Type: "str", + Required: false, + }}, + }}}, + } + + require.NoError(t, validateManifest(nil)) + require.NoError(t, validateManifest(valid)) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := validateManifest(tt.manifest) + require.Error(t, err) + require.ErrorIs(t, err, errInvalidManifest) + }) + } +} + +// TestFrameLimitsCheckedCaps proves payload caps fit uint32 and follow envelope math. +func TestFrameLimitsCheckedCaps(t *testing.T) { + manifest := validManifest() + maxValueBytes := 64 + + childCap, err := childPayloadCap(maxValueBytes, manifest) + require.NoError(t, err) + parentCap, err := parentPayloadCap(maxValueBytes) + require.NoError(t, err) + execCap, err := execPayloadCap(32, manifest) + require.NoError(t, err) + + abort, err := encodeNativeAbort() + require.NoError(t, err) + finalError, err := encodeFinalError(finalErrorInvalidArguments) + require.NoError(t, err) + assert.GreaterOrEqual(t, childCap, uint32(len(finalError))) + assert.GreaterOrEqual(t, parentCap, uint32(len(abort))) + assert.Greater(t, childCap, uint32(maxValueBytes)) + assert.Greater(t, parentCap, uint32(maxValueBytes)) + assert.Greater(t, execCap, uint32(len(manifestJSON(t, manifest)))) + + longer := []manifestEntry{{ + ID: strings.Repeat("a", 40), + Name: "records.lookup", + Input: manifest[0].Input, + }} + require.NoError(t, validateManifest(longer)) + longerCap, err := childPayloadCap(maxValueBytes, longer) + require.NoError(t, err) + assert.Greater(t, longerCap, childCap) + + escaped := []manifestEntry{{ + ID: strings.Repeat("\"", 40), + Name: "records.lookup", + Input: manifest[0].Input, + }} + require.NoError(t, validateManifest(escaped)) + escapedCap, err := childPayloadCap(maxValueBytes, escaped) + require.NoError(t, err) + assert.Greater(t, escapedCap, longerCap) + + expanded, err := execPayloadCap(8, manifest) + require.NoError(t, err) + unexpanded, err := execPayloadCap(1, manifest) + require.NoError(t, err) + assert.Greater(t, expanded, unexpanded) +} + +// TestFrameLimitsRejectUint32Overflow proves checked addition never wraps the prefix. +func TestFrameLimitsRejectUint32Overflow(t *testing.T) { + manifest := validManifest() + + _, err := childPayloadCap(math.MaxInt, manifest) + require.Error(t, err) + require.ErrorIs(t, err, errInvalidLimits) + + _, err = parentPayloadCap(math.MaxInt) + require.Error(t, err) + require.ErrorIs(t, err, errInvalidLimits) + + _, err = execPayloadCap(math.MaxInt, manifest) + require.Error(t, err) + require.ErrorIs(t, err, errInvalidLimits) +} + +// TestFrameLimitsOversizedValueIsProtocolViolation proves over-cap values are not resource-limit reports. +func TestFrameLimitsOversizedValueIsProtocolViolation(t *testing.T) { + exec := validExecFrame() + exec.Limits.MaxValueBytes = 8 + parent, child := newBufferedExecPair(t, exec) + require.NoError(t, parent.writeExec(exec)) + _, err := child.read() + require.NoError(t, err) + + oversized := strings.Repeat("a", 32) + err = child.writeFinal(oversized) + require.Error(t, err) + assert.True(t, errors.Is(err, errInvalidValue) || errors.Is(err, errFrameTooLarge)) + assert.NotErrorIs(t, err, errIllegalState) +} + +// TestFrameDecodeNormalizesNestedValues proves UseNumber results never escape decodePayload. +func TestFrameDecodeNormalizesNestedValues(t *testing.T) { + payload := `{"type":"final","result":{"count":1,"score":1.0,"items":[2,3.0]}}` + decoded, err := decodeFramed(t, payload) + require.NoError(t, err) + frame, ok := decoded.(finalFrame) + require.True(t, ok) + result, ok := frame.Result.(map[string]any) + require.True(t, ok) + assert.IsType(t, int64(0), result["count"]) + assert.IsType(t, float64(0), result["score"]) + items, ok := result["items"].([]any) + require.True(t, ok) + assert.IsType(t, int64(0), items[0]) + assert.IsType(t, float64(0), items[1]) +} + +func validExecFrame() execFrame { + return execFrame{ + Type: frameTypeExec, + Version: protocolVersion, + Source: "def main():\n return 1\n", + Limits: childLimits{ + MaxSourceBytes: 256, + MaxExecutionSteps: 1000, + MaxNativeCalls: 8, + MaxValueDepth: 8, + MaxValueBytes: 256, + }, + Manifest: validManifest(), + } +} + +func validManifest() []manifestEntry { + return []manifestEntry{{ + ID: "cap.lookup", + Name: "records.lookup", + Input: []binding.FieldShape{ + {Name: "org", Type: "str", Required: true}, + {Name: "limit", Type: "int | None", Required: false}, + }, + }} +} + +func newProbePair(t *testing.T) (*parentConn, *childConn, func()) { + t.Helper() + childReader, parentWriter := io.Pipe() + parentReader, childWriter := io.Pipe() + closePipes := func() { + _ = childReader.Close() + _ = parentWriter.Close() + _ = parentReader.Close() + _ = childWriter.Close() + } + t.Cleanup(closePipes) + return newParentProbeConn(parentWriter, parentReader), newChildProbeConn(childReader, childWriter), closePipes +} + +func newExecPair(t *testing.T, exec execFrame) (*parentConn, *childConn, func()) { + t.Helper() + execCap, err := execPayloadCap(exec.Limits.MaxSourceBytes, exec.Manifest) + require.NoError(t, err) + parentCap, err := parentPayloadCap(exec.Limits.MaxValueBytes) + require.NoError(t, err) + childCap, err := childPayloadCap(exec.Limits.MaxValueBytes, exec.Manifest) + require.NoError(t, err) + + childReader, parentWriter := io.Pipe() + parentReader, childWriter := io.Pipe() + closePipes := func() { + _ = childReader.Close() + _ = parentWriter.Close() + _ = parentReader.Close() + _ = childWriter.Close() + } + t.Cleanup(closePipes) + return newParentExecConn(parentWriter, parentReader, childCap, parentCap), + newChildExecConn(childReader, childWriter, execCap, parentCap, childCap), + closePipes +} + +func decodeFramed(t *testing.T, payload string) (any, error) { + t.Helper() + var buf bytes.Buffer + require.NoError(t, writeFrame(&buf, []byte(payload), testReadCap)) + raw, err := readFrame(&buf, testReadCap) + if err != nil { + return nil, err + } + return decodePayload(raw) +} + +func newBufferedExecPair(t *testing.T, exec execFrame) (*parentConn, *childConn) { + t.Helper() + execCap, err := execPayloadCap(exec.Limits.MaxSourceBytes, exec.Manifest) + require.NoError(t, err) + parentCap, err := parentPayloadCap(exec.Limits.MaxValueBytes) + require.NoError(t, err) + childCap, err := childPayloadCap(exec.Limits.MaxValueBytes, exec.Manifest) + require.NoError(t, err) + + var toChild bytes.Buffer + var toParent bytes.Buffer + return newParentExecConn(&toChild, &toParent, childCap, parentCap), + newChildExecConn(&toChild, &toParent, execCap, parentCap, childCap) +} + +func prefixed(t *testing.T, payload []byte) []byte { + t.Helper() + return append(u32(uint32(len(payload))), payload...) +} + +func u32(value uint32) []byte { + raw := make([]byte, 4) + binary.BigEndian.PutUint32(raw, value) + return raw +} + +func assertFrameType(t *testing.T, frame any, want string) { + t.Helper() + switch got := frame.(type) { + case probeFrame: + assert.Equal(t, want, got.Type) + case probeAckFrame: + assert.Equal(t, want, got.Type) + case execFrame: + assert.Equal(t, want, got.Type) + case nativeCallFrame: + assert.Equal(t, want, got.Type) + case nativeResultFrame: + assert.Equal(t, want, got.Type) + case nativeAbortFrame: + assert.Equal(t, want, got.Type) + case finalFrame: + assert.Equal(t, want, got.Type) + case finalErrorFrame: + assert.Equal(t, want, got.Type) + default: + t.Fatalf("unexpected frame type %T", frame) + } +} + +func manifestJSON(t *testing.T, manifest []manifestEntry) []byte { + t.Helper() + encoded, err := json.Marshal(manifest) + require.NoError(t, err) + return encoded +} diff --git a/internal/worker/limits.go b/internal/worker/limits.go new file mode 100644 index 0000000..f1d957c --- /dev/null +++ b/internal/worker/limits.go @@ -0,0 +1,286 @@ +package worker + +import ( + "encoding/json" + "fmt" + "math" + "strings" + + "github.com/meigma/codemode/internal/binding" +) + +const ( + minimumDottedSegments = 2 + jsonStringEscapeMax = 6 +) + +// childLimits contains the child-visible positive execution budgets. +type childLimits struct { + // MaxSourceBytes bounds submitted source length. + MaxSourceBytes int `json:"max_source_bytes"` + + // MaxExecutionSteps bounds interpreter work. + MaxExecutionSteps uint64 `json:"max_execution_steps"` + + // MaxNativeCalls bounds attempted native invocations. + MaxNativeCalls uint64 `json:"max_native_calls"` + + // MaxValueDepth bounds recursive converted-value nesting. + MaxValueDepth int `json:"max_value_depth"` + + // MaxValueBytes bounds type-preserving encoded value size. + MaxValueBytes int `json:"max_value_bytes"` +} + +// manifestEntry is one enabled capability advertised to the child. +type manifestEntry struct { + // ID is the stable capability identity. + ID string `json:"id"` + + // Name is the dotted Starlark path. + Name string `json:"name"` + + // Input is the exact compiled input shape. + Input []binding.FieldShape `json:"input"` +} + +// childPayloadCap is the largest legal child-originated execution payload. +func childPayloadCap(maxValueBytes int, manifest []manifestEntry) (uint32, error) { + valueBytes, err := configUint32(maxValueBytes) + if err != nil { + return 0, err + } + longestID, err := longestEncodedID(manifest) + if err != nil { + return 0, err + } + nativeCall, err := addConfigUint32( + uint32(len(nativeCallPrefix)), + longestID, + uint32(len(nativeCallMid)), + valueBytes, + uint32(len(nativeCallSuffix)), + ) + if err != nil { + return 0, err + } + final, err := addConfigUint32(uint32(len(finalPrefix)), valueBytes, uint32(len(finalSuffix))) + if err != nil { + return 0, err + } + return maxUint32(nativeCall, final, uint32(len(finalErrorLongest))), nil +} + +// parentPayloadCap is the largest legal parent-originated execution payload. +func parentPayloadCap(maxValueBytes int) (uint32, error) { + valueBytes, err := configUint32(maxValueBytes) + if err != nil { + return 0, err + } + nativeResult, err := addConfigUint32(uint32(len(nativeResultPrefix)), valueBytes, uint32(len(nativeResultSuffix))) + if err != nil { + return 0, err + } + return maxUint32(nativeResult, uint32(len(nativeAbortPayload))), nil +} + +// execPayloadCap is the largest legal initial exec payload. +func execPayloadCap(maxSourceBytes int, manifest []manifestEntry) (uint32, error) { + if maxSourceBytes < 0 { + return 0, errInvalidLimits + } + base, err := marshalFrame(execFrame{ + Type: frameTypeExec, + Version: protocolVersion, + Source: "", + Limits: worstChildLimits(), + Manifest: emptyManifest(manifest), + }) + if err != nil { + return 0, err + } + sourceBytes, err := configUint32(maxSourceBytes) + if err != nil { + return 0, err + } + expanded, err := mulConfigUint32(jsonStringEscapeMax, sourceBytes) + if err != nil { + return 0, err + } + baseLen, err := configUint32(len(base)) + if err != nil { + return 0, err + } + return addConfigUint32(baseLen, expanded) +} + +// validateChildLimits reports whether every child budget is positive. +func validateChildLimits(limits childLimits) error { + if limits.MaxSourceBytes <= 0 || + limits.MaxExecutionSteps == 0 || + limits.MaxNativeCalls == 0 || + limits.MaxValueDepth <= 0 || + limits.MaxValueBytes <= 0 { + return errInvalidLimits + } + return nil +} + +// validateManifest reports whether capability identities and shapes are legal. +func validateManifest(entries []manifestEntry) error { + ids := make(map[string]struct{}, len(entries)) + names := make(map[string]struct{}, len(entries)) + for _, entry := range entries { + if entry.ID == "" || entry.ID != strings.TrimSpace(entry.ID) { + return errInvalidManifest + } + if _, duplicate := ids[entry.ID]; duplicate { + return errInvalidManifest + } + if !isDottedName(entry.Name) { + return errInvalidManifest + } + if _, duplicate := names[entry.Name]; duplicate { + return errInvalidManifest + } + if err := binding.ValidateInputShape(entry.Input); err != nil { + return fmt.Errorf("%w: %w", errInvalidManifest, err) + } + ids[entry.ID] = struct{}{} + names[entry.Name] = struct{}{} + } + return validateNamespaceCollisions(names) +} + +// isDottedName reports whether name has at least two valid Starlark segments. +func isDottedName(name string) bool { + segments := strings.Split(name, ".") + if len(segments) < minimumDottedSegments { + return false + } + for _, segment := range segments { + if !binding.ValidIdentifier(segment) { + return false + } + } + return true +} + +// validateNamespaceCollisions rejects a capability that is also a namespace. +func validateNamespaceCollisions(names map[string]struct{}) error { + for name := range names { + segments := strings.Split(name, ".") + for end := 1; end < len(segments); end++ { + prefix := strings.Join(segments[:end], ".") + if _, collides := names[prefix]; collides { + return errInvalidManifest + } + } + } + return nil +} + +// longestEncodedID returns the longest JSON-encoded capability ID. +func longestEncodedID(manifest []manifestEntry) (uint32, error) { + longest := uint32(len(emptyJSONString)) + for _, entry := range manifest { + encoded, err := json.Marshal(entry.ID) + if err != nil { + return 0, errInvalidManifest + } + length, err := configUint32(len(encoded)) + if err != nil { + return 0, err + } + if length > longest { + longest = length + } + } + return longest, nil +} + +// emptyManifest returns a non-nil copy used for exec cap encoding. +func emptyManifest(manifest []manifestEntry) []manifestEntry { + if manifest == nil { + return []manifestEntry{} + } + return manifest +} + +// worstChildLimits is the largest numeric spelling used for exec cap math. +func worstChildLimits() childLimits { + return childLimits{ + MaxSourceBytes: math.MaxInt, + MaxExecutionSteps: math.MaxUint64, + MaxNativeCalls: math.MaxUint64, + MaxValueDepth: math.MaxInt, + MaxValueBytes: math.MaxInt, + } +} + +// uint32FromInt converts a non-negative int that fits a uint32 prefix. +func uint32FromInt(value int) (uint32, error) { + if value < 0 || uint64(value) > math.MaxUint32 { + return 0, errFrameTooLarge + } + return uint32(value), nil +} + +// configUint32 converts a configured size that must fit the uint32 prefix. +func configUint32(value int) (uint32, error) { + converted, err := uint32FromInt(value) + if err != nil { + return 0, errInvalidLimits + } + return converted, nil +} + +// addUint32 adds values and rejects unsigned 32-bit overflow. +func addUint32(values ...uint32) (uint32, error) { + var sum uint64 + for _, value := range values { + sum += uint64(value) + if sum > math.MaxUint32 { + return 0, errFrameTooLarge + } + } + return uint32(sum), nil +} + +// mulUint32 multiplies two values and rejects unsigned 32-bit overflow. +func mulUint32(left uint32, right uint32) (uint32, error) { + product := uint64(left) * uint64(right) + if product > math.MaxUint32 { + return 0, errFrameTooLarge + } + return uint32(product), nil +} + +// addConfigUint32 adds configured sizes and rejects prefix overflow as invalid limits. +func addConfigUint32(values ...uint32) (uint32, error) { + sum, err := addUint32(values...) + if err != nil { + return 0, errInvalidLimits + } + return sum, nil +} + +// mulConfigUint32 multiplies configured sizes and rejects prefix overflow as invalid limits. +func mulConfigUint32(left uint32, right uint32) (uint32, error) { + product, err := mulUint32(left, right) + if err != nil { + return 0, errInvalidLimits + } + return product, nil +} + +// maxUint32 returns the largest supplied value. +func maxUint32(values ...uint32) uint32 { + var largest uint32 + for _, value := range values { + if value > largest { + largest = value + } + } + return largest +} diff --git a/internal/worker/value.go b/internal/worker/value.go new file mode 100644 index 0000000..0708822 --- /dev/null +++ b/internal/worker/value.go @@ -0,0 +1,195 @@ +package worker + +import ( + "bytes" + "encoding/json" + "fmt" + "math" + "slices" + "strconv" + "strings" + "unicode/utf8" + + "github.com/meigma/codemode/internal/binding" +) + +// encodeNormalizedValue encodes one process-neutral value with sorted keys. +func encodeNormalizedValue(value any) ([]byte, error) { + var buf bytes.Buffer + if err := writeNormalizedValue(&buf, value); err != nil { + return nil, err + } + return buf.Bytes(), nil +} + +// decodeNormalizedValue converts one JSON value into the normalized domain. +func decodeNormalizedValue(raw json.RawMessage) (any, error) { + if len(raw) == 0 { + return nil, errMalformedJSON + } + if !utf8.Valid(raw) { + return nil, errInvalidUTF8 + } + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.UseNumber() + var value any + if err := dec.Decode(&value); err != nil { + return nil, classifyJSONErr(err) + } + if err := requireEOF(dec); err != nil { + return nil, err + } + return normalizeValue(value) +} + +// writeNormalizedValue writes one supported value in canonical JSON form. +func writeNormalizedValue(buf *bytes.Buffer, value any) error { + switch typed := value.(type) { + case nil: + buf.WriteString("null") + return nil + case bool: + if typed { + buf.WriteString("true") + return nil + } + buf.WriteString("false") + return nil + case string: + encoded, err := json.Marshal(typed) + if err != nil { + return errInvalidValue + } + buf.Write(encoded) + return nil + case int64: + buf.WriteString(strconv.FormatInt(typed, 10)) + return nil + case float64: + return writeFloat(buf, typed) + case []any: + return writeNormalizedList(buf, typed) + case map[string]any: + return writeNormalizedObject(buf, typed) + default: + return errInvalidValue + } +} + +// writeNormalizedList writes one process-neutral list. +func writeNormalizedList(buf *bytes.Buffer, value []any) error { + buf.WriteByte('[') + for index, item := range value { + if index > 0 { + buf.WriteByte(',') + } + if err := writeNormalizedValue(buf, item); err != nil { + return err + } + } + buf.WriteByte(']') + return nil +} + +// writeNormalizedObject writes one process-neutral object with sorted keys. +func writeNormalizedObject(buf *bytes.Buffer, value map[string]any) error { + keys := make([]string, 0, len(value)) + for key := range value { + keys = append(keys, key) + } + slices.Sort(keys) + buf.WriteByte('{') + for index, key := range keys { + if index > 0 { + buf.WriteByte(',') + } + encoded, err := json.Marshal(key) + if err != nil { + return errInvalidValue + } + buf.Write(encoded) + buf.WriteByte(':') + if err := writeNormalizedValue(buf, value[key]); err != nil { + return err + } + } + buf.WriteByte('}') + return nil +} + +// writeFloat writes one finite float in shortest round-trippable spelling. +func writeFloat(buf *bytes.Buffer, value float64) error { + if math.IsNaN(value) || math.IsInf(value, 0) { + return errInvalidNumber + } + encoded := strconv.FormatFloat(value, 'g', -1, 64) + if !strings.ContainsAny(encoded, ".eE") { + encoded += ".0" + } + buf.WriteString(encoded) + return nil +} + +// normalizeValue converts decoded JSON into the process-neutral domain. +func normalizeValue(value any) (any, error) { + switch typed := value.(type) { + case nil, bool, string: + return typed, nil + case json.Number: + return decodeNumber(typed) + case []any: + normalized := make([]any, len(typed)) + for index, item := range typed { + converted, err := normalizeValue(item) + if err != nil { + return nil, err + } + normalized[index] = converted + } + return normalized, nil + case map[string]any: + normalized := make(map[string]any, len(typed)) + for key, item := range typed { + converted, err := normalizeValue(item) + if err != nil { + return nil, err + } + normalized[key] = converted + } + return normalized, nil + default: + return nil, errInvalidValue + } +} + +// decodeNumber preserves int64 tokens and finite fractional or exponent floats. +func decodeNumber(number json.Number) (any, error) { + token := number.String() + if strings.ContainsAny(token, ".eE") { + parsed, err := strconv.ParseFloat(token, 64) + if err != nil || math.IsNaN(parsed) || math.IsInf(parsed, 0) { + return nil, errInvalidNumber + } + return parsed, nil + } + parsed, err := strconv.ParseInt(token, 10, 64) + if err != nil { + return nil, errInvalidNumber + } + return parsed, nil +} + +// validateBoundedValue enforces the shared type, depth, and encoded-size limits. +func validateBoundedValue(value any, limits childLimits) error { + if err := binding.ValidateValue(value, limits.MaxValueDepth, limits.MaxValueBytes); err != nil { + return fmt.Errorf("%w: %w", errInvalidValue, err) + } + encoded, err := encodeNormalizedValue(value) + if err != nil { + return err + } + if len(encoded) > limits.MaxValueBytes { + return errFrameTooLarge + } + return nil +} diff --git a/internal/worker/value_test.go b/internal/worker/value_test.go new file mode 100644 index 0000000..c6504d3 --- /dev/null +++ b/internal/worker/value_test.go @@ -0,0 +1,170 @@ +package worker + +import ( + "bytes" + "encoding/json" + "errors" + "math" + "strconv" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/meigma/codemode/internal/binding" +) + +// TestValueCodecPreservesNumericIdentity proves tokens keep int64 and finite float64 apart. +func TestValueCodecPreservesNumericIdentity(t *testing.T) { + negativeZero := math.Copysign(0, -1) + + tests := []struct { + // name identifies the numeric identity. + name string + + // value is the normalized Go value. + value any + + // json is the exact type-preserving spelling. + json string + }{ + {name: "minimum int64", value: int64(math.MinInt64), json: "-9223372036854775808"}, + {name: "maximum int64", value: int64(math.MaxInt64), json: "9223372036854775807"}, + {name: "integer zero", value: int64(0), json: "0"}, + {name: "integral float", value: 1.0, json: "1.0"}, + {name: "fractional float", value: 1.25, json: "1.25"}, + {name: "exponent float", value: 1e20, json: strconv.FormatFloat(1e20, 'g', -1, 64)}, + {name: "negative zero", value: negativeZero, json: "-0.0"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + encoded, err := encodeNormalizedValue(tt.value) + require.NoError(t, err) + assert.Equal(t, tt.json, string(encoded)) + + decoded, err := decodeNormalizedValue(json.RawMessage(encoded)) + require.NoError(t, err) + assert.Equal(t, tt.value, decoded) + assert.IsType(t, tt.value, decoded) + if tt.name == "negative zero" { + got, ok := decoded.(float64) + require.True(t, ok) + assert.True(t, math.Signbit(got)) + } + }) + } +} + +// TestValueCodecDecodesTokensBySpelling proves decoder rules follow the token, not Go defaults. +func TestValueCodecDecodesTokensBySpelling(t *testing.T) { + tests := []struct { + // name identifies the token class. + name string + + // token is the raw JSON number. + token string + + // want is the normalized value. + want any + }{ + {name: "integer token", token: "7", want: int64(7)}, + {name: "fraction token", token: "1.0", want: 1.0}, + {name: "exponent token", token: "1e2", want: 100.0}, + {name: "negative exponent", token: "-2.5E+1", want: -25.0}, + {name: "negative zero token", token: "-0.0", want: math.Copysign(0, -1)}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := decodeNormalizedValue(json.RawMessage(tt.token)) + require.NoError(t, err) + assert.Equal(t, tt.want, got) + assert.IsType(t, tt.want, got) + _, isNumber := got.(json.Number) + assert.False(t, isNumber, "json.Number must not escape the codec") + }) + } +} + +// TestValueCodecSortsObjectKeys proves maps have a deterministic wire order. +func TestValueCodecSortsObjectKeys(t *testing.T) { + encoded, err := encodeNormalizedValue(map[string]any{ + "zeta": int64(1), + "alpha": map[string]any{ + "n": int64(2), + "a": true, + }, + }) + require.NoError(t, err) + assert.True(t, bytes.Equal(encoded, []byte(`{"alpha":{"a":true,"n":2},"zeta":1}`))) +} + +// TestValueCodecRejectsUnsupportedValues proves [json.Number] never leaves the codec. +func TestValueCodecRejectsUnsupportedValues(t *testing.T) { + encodeTests := []struct { + // name identifies the rejected Go value. + name string + + // value is presented to the encoder. + value any + + // target is the expected protocol error. + target error + }{ + {name: "json.Number", value: json.Number("1"), target: errInvalidValue}, + {name: "int", value: 1, target: errInvalidValue}, + {name: "uint64", value: uint64(1), target: errInvalidValue}, + {name: "float32", value: float32(1), target: errInvalidValue}, + {name: "NaN", value: math.NaN(), target: errInvalidNumber}, + {name: "infinity", value: math.Inf(1), target: errInvalidNumber}, + {name: "negative infinity", value: math.Inf(-1), target: errInvalidNumber}, + {name: "string map", value: map[string]string{"k": "v"}, target: errInvalidValue}, + } + + for _, tt := range encodeTests { + t.Run("encode "+tt.name, func(t *testing.T) { + _, err := encodeNormalizedValue(tt.value) + require.Error(t, err) + require.ErrorIs(t, err, tt.target) + }) + } + + decodeTests := []struct { + // name identifies the rejected token. + name string + + // token is raw JSON. + token string + }{ + {name: "integer overflow", token: "9223372036854775808"}, + {name: "integer underflow", token: "-9223372036854775809"}, + {name: "NaN token", token: "NaN"}, + {name: "Infinity token", token: "Infinity"}, + {name: "non-string key", token: `{1:true}`}, + } + + for _, tt := range decodeTests { + t.Run("decode "+tt.name, func(t *testing.T) { + _, err := decodeNormalizedValue(json.RawMessage(tt.token)) + require.Error(t, err) + assert.True( + t, + errors.Is(err, errInvalidNumber) || errors.Is(err, errInvalidValue) || errors.Is(err, errMalformedJSON), + ) + }) + } +} + +// TestValueCodecUsesBindingValidation proves value domain checks are not reimplemented here. +func TestValueCodecUsesBindingValidation(t *testing.T) { + value := map[string]any{"org": "meigma", "limit": int64(25), "nested": []any{nil, true, 1.0}} + require.NoError(t, binding.ValidateValue(value, 4, 1024)) + + encoded, err := encodeNormalizedValue(value) + require.NoError(t, err) + decoded, err := decodeNormalizedValue(encoded) + require.NoError(t, err) + assert.Equal(t, value, decoded) + require.NoError(t, binding.ValidateValue(decoded, 4, 1024)) +}