<commit_before>// Package mqtt implements the Message Queuing Telemetry Transport (MQTT)
// protocol.
//
// http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html
package mqtt

import (
	"errors"
	"fmt"
	"net"
	"sync"
)

// Fixed Packets
var (
	pingPacket    = []byte{ping << 4, 0}
	pongPacket    = []byte{pong << 4, 0}
	disconnPacket = []byte{disconn << 4, 0}
)

const (
	packetMax = 268_435_455 // 4-byte varint
	stringMax = 65535       // 16-bit size prefixes
)

var (
	errPacketMax = errors.New("mqtt: packet limit of 256 MiB exceeded")
	errStringMax = errors.New("mqtt: string exceeds 65535 bytes")

	errUTF8 = errors.New("mqtt: invalid UTF-8 byte sequence")
	errNull = errors.New("mqtt: string contains null characer")
)

func stringCheck(s string) error {
	if len(s) > stringMax {
		return errStringMax
	}
	for _, r := range s {
		// “The character data in a UTF-8 encoded string MUST be
		// well-formed UTF-8 as defined by the Unicode specification …”
		// — MQTT Version 3.1.1, conformance statement MQTT-1.5.3-1
		if r == '\uFFFD' {
			return errUTF8
		}

		// “A UTF-8 encoded string MUST NOT include an encoding of the
		// null character U+0000.”
		// — MQTT Version 3.1.1, conformance statement MQTT-1.5.3-2
		if r == 0 {
			return errNull
		}
	}
	return nil
}

// ErrRequestLimit signals too many client requests.
var ErrRequestLimit = errors.New("mqtt: maximum number of pending requests reached")

// Protocol Constraints
var (
	ErrTopicName   = errors.New("mqtt: malformed topic name: null [U+0000] character, illegal UTF-8 sequence or size exceeds 64 kiB")
	ErrMessageSize = errors.New("mqtt: message size exceeds 268,435,451 B minus UTF-8 length of topic name")
)

// Quality of Service (QoS) are defined as numeric levels.
const (
	atMostOnce  = iota // fire and forget
	atLeastOnce        // network round trip + Persistence
	exactlyOnce        // two network round trips + Persistence
)

// Control Packet Types
const (
	connReq = iota + 1
	connAck
	pubMsg
	pubAck      // QoS 1 confirm
	pubReceived // QoS 2 confirm, part Ⅰ
	pubRelease  // QoS 2 confirm, part Ⅱ
	pubComplete // QoS 2 confirm, part Ⅲ
	subReq
	subAck
	unsubReq
	unsubAck
	ping
	pong
	disconn
)

// Publication Flags
const (
	// Indicates that this might be re-delivery of an earlier attempt.
	dupeFlag = 8
	// Store the application message and its QoS, so that it can be
	// delivered to future subscribers.
	retainFlag = 1
)

// SessionConfig defines protocol settings.
type SessionConfig struct {
	// Required for the server to uniquely identify the session.
	ClientID string

	// Optional identification claim.
	UserName string
	// Optional credentials.
	Password []byte

	// Discard any previous session (on the server) and start a new one.
	CleanSession bool

	// Will is a message publication to be send when the
	// connection terminates without a disconnect package.
	Will struct {
		// Destination for the message.
		Topic string

		// The Will feature is disabled when nil.
		Message []byte

		// Persistence constraints.
		Retain bool

		// Delivery constraints.
		AtLeastOnce bool // raises the QoS to level 1
		ExactlyOnce bool // raises the QoS to level 2
	}

	// Timeout in seconds or disabled when zero.
	KeepAlive uint16
}

// Valid returns a protocol constraint violation, if any.
func (c *SessionConfig) Valid() error {
	if err := stringCheck(c.ClientID); err != nil {
		return fmt.Errorf("mqtt: illegal client identifier: %w", err)
	}
	if err := stringCheck(c.UserName); err != nil {
		return fmt.Errorf("mqtt: illegal user name: %w", err)
	}
	if len(c.Password) > stringMax {
		return fmt.Errorf("mqtt: password exceeds %d bytes", stringMax)
	}
	if c.Will.Message != nil {
		// MQTT Version 3.1.1, conformance statement MQTT-3.1.3-10
		if err := stringCheck(c.Will.Topic); err != nil {
			return fmt.Errorf("mqtt: illegal will topic: %w", err)
		}
		if len(c.Will.Message) > stringMax {
			return fmt.Errorf("mqtt: will message exceeds %d bytes", stringMax)
		}
	}
	return nil
}

func (c *SessionConfig) appendConnReq(dst []byte) []byte {
	size := 6 + 1 + 1 + 2 // protocol name + protocol level + connect flags + keep alive
	size += 2 + len(c.ClientID)
	var flags uint
	if c.UserName != "" {
		size += 2 + len(c.UserName)
		flags |= 1 << 7
	}
	if c.Password != nil {
		size += 2 + len(c.Password)
		flags |= 1 << 6
	}
	if c.Will.Message != nil {
		size += 4 + len(c.Will.Topic) + len(c.Will.Message)
		if c.Will.Retain {
			flags |= 1 << 5
		}
		switch {
		case c.Will.ExactlyOnce:
			flags |= exactlyOnce << 3
		case c.Will.AtLeastOnce:
			flags |= atLeastOnce << 3
		}
		flags |= 1 << 2
	}<commit_msg>Validation polish.<commit_after>// Package mqtt implements the Message Queuing Telemetry Transport (MQTT)
// protocol.
//
// http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html
package mqtt

import (
	"errors"
	"fmt"
	"net"
	"sync"
)

// Fixed Packets
var (
	pingPacket    = []byte{ping << 4, 0}
	pongPacket    = []byte{pong << 4, 0}
	disconnPacket = []byte{disconn << 4, 0}
)

const (
	// See MQTT Version 3.1.1, table 2.4: “Size of Remaining Length field”.
	packetMax = 1<<(4*7) - 1 // 4-byte varint

	// “Unless stated otherwise all UTF-8 encoded strings can have any
	// length in the range 0 to 65535 bytes.”
	// — MQTT Version 3.1.1, section 1.5.3
	stringMax = 1<<16 - 1 // 16-bit size prefixes
)

// Validation errors are expected to be prefixed according to context.
var (
	// ErrPacketMax enforces packetMax.
	errPacketMax = errors.New("packet payload exceeds 256 MiB")
	// ErrStringMax enforces stringMax.
	errStringMax = errors.New("string exceeds 64 KiB")

	errUTF8 = errors.New("invalid UTF-8 byte sequence")
	errNull = errors.New("string contains null characer")
)

func stringCheck(s string) error {
	if len(s) > stringMax {
		return errStringMax
	}
	for _, r := range s {
		// “The character data in a UTF-8 encoded string MUST be
		// well-formed UTF-8 as defined by the Unicode specification
		// and restated in RFC 3629.”
		// — MQTT Version 3.1.1, conformance statement MQTT-1.5.3-1
		if r == '\uFFFD' {
			return errUTF8
		}

		// “A UTF-8 encoded string MUST NOT include an encoding of the
		// null character U+0000.”
		// — MQTT Version 3.1.1, conformance statement MQTT-1.5.3-2
		if r == 0 {
			return errNull
		}
	}
	return nil
}

// ErrRequestLimit signals too many client requests.
var ErrRequestLimit = errors.New("mqtt: maximum number of pending requests reached")

// Quality of Service (QoS) are defined as numeric levels.
const (
	atMostOnce  = iota // fire and forget
	atLeastOnce        // network round trip + Persistence
	exactlyOnce        // two network round trips + Persistence
)

// Control Packet Types
const (
	connReq = iota + 1
	connAck
	pubMsg
	pubAck      // QoS 1 confirm
	pubReceived // QoS 2 confirm, part Ⅰ
	pubRelease  // QoS 2 confirm, part Ⅱ
	pubComplete // QoS 2 confirm, part Ⅲ
	subReq
	subAck
	unsubReq
	unsubAck
	ping
	pong
	disconn
)

// Publication Flags
const (
	// Indicates that this might be re-delivery of an earlier attempt.
	dupeFlag = 8
	// Store the application message and its QoS, so that it can be
	// delivered to future subscribers.
	retainFlag = 1
)

// SessionConfig defines protocol settings.
type SessionConfig struct {
	// Required for the server to uniquely identify the session.
	ClientID string

	// Optional identification claim.
	UserName string
	// Optional credentials.
	Password []byte

	// Discard any previous session (on the server) and start a new one.
	// Note that automated reconnects will use false only.
	CleanSession bool

	// Will is a message publication to be send when the
	// connection terminates without a disconnect package.
	Will struct {
		// Destination for the message.
		Topic string

		// The Will feature is disabled when nil.
		Message []byte

		// Persistence constraints.
		Retain bool

		// Delivery constraints.
		AtLeastOnce bool // raises the QoS to level 1
		ExactlyOnce bool // raises the QoS to level 2
	}

	// Timeout in seconds or disabled when zero.
	KeepAlive uint16
}

// Valid returns a protocol constraint violation, if any.
func (c *SessionConfig) Valid() error {
	if err := stringCheck(c.ClientID); err != nil {
		return fmt.Errorf("mqtt: illegal client identifier: %w", err)
	}
	if err := stringCheck(c.UserName); err != nil {
		return fmt.Errorf("mqtt: illegal user name: %w", err)
	}
	if len(c.Password) > stringMax {
		return fmt.Errorf("mqtt: password exceeds %d bytes", stringMax)
	}
	if c.Will.Message != nil {
		// MQTT Version 3.1.1, conformance statement MQTT-3.1.3-10
		if err := stringCheck(c.Will.Topic); err != nil {
			return fmt.Errorf("mqtt: illegal will topic: %w", err)
		}
		if len(c.Will.Message) > stringMax {
			return fmt.Errorf("mqtt: will message exceeds %d bytes", stringMax)
		}
	}
	return nil
}

// AppendConnReq appends the CONNREQ packet to dst and returns the extended buffer.
func (c *SessionConfig) appendConnReq(dst []byte) []byte {
	size := 2 + 4 // protocol name
	size += 1     // protocol level
	size += 1     // connect flags
	size += 2     // keep alive
	size += 2 + len(c.ClientID)
	var flags uint
	if c.UserName != "" {
		size += 2 + len(c.UserName)
		flags |= 1 << 7
	}
	if c.Password != nil {
		size += 2 + len(c.Password)
		flags |= 1 << 6
	}
	if c.Will.Message != nil {
		size += 4 + len(c.Will.Topic) + len(c.Will.Message)
		if c.Will.Retain {
			flags |= 1 << 5
		}
		switch {
		case c.Will.ExactlyOnce:
			flags |= exactlyOnce << 3
		case c.Will.AtLeastOnce:
			flags |= atLeastOnce << 3
		}
		flags |= 1 << 2
	}
{'fraction_non_alphanumeric': 0.11356963087248322, 'fraction_numerical': 0.02401426174496644, 'mean_word_length': 3.025749261291684, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1801066', 'n_tokens_mistral': 3536, 'n_tokens_neox': 3179, 'n_words': 1127}