stream

package
v0.5.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 12, 2026 License: Apache-2.0 Imports: 13 Imported by: 1

Documentation

Overview

Package stream provides low-level TCP socket stream management and message framing for the CEDAR protocol.

This package implements the binary framing protocol that wraps messages before sending them over TCP sockets, based on HTCondor's reli_sock.cpp implementation.

Index

Constants

View Source 
const (
	// DefaultKeepAliveIdle is the idle time before the first keepalive probe.
	// Matches TCP_KEEPALIVE_INTERVAL's default of 360 seconds.
	DefaultKeepAliveIdle = 360 * time.Second
	// DefaultKeepAliveInterval is the interval between probes (C++ TCP_KEEPINTVL = 5s).
	DefaultKeepAliveInterval = 5 * time.Second
	// DefaultKeepAliveCount is the number of failed probes before the
	// connection is declared dead (C++ TCP_KEEPCNT = 5).
	DefaultKeepAliveCount = 5
)

TCP keepalive defaults for CEDAR sockets.

These mirror C++ HTCondor's Sock::set_keepalive() (src/condor_io/sock.cpp:955), which is driven by the TCP_KEEPALIVE_INTERVAL config knob. That knob defaults to 360 (src/condor_utils/param_info.in:4587-4589) and is interpreted as the idle time (seconds) before the first probe; the C++ code then hard-codes a probe count of 5 (TCP_KEEPCNT) and a probe interval of 5 seconds (TCP_KEEPINTVL). HTCondor enables SO_KEEPALIVE on outbound (client-dialed) TCP sockets and on server-accepted sockets so a silently-dead peer (e.g. an execute host that lost power) is detected instead of leaving a goroutine blocked in Read forever.

Semantics of the knob, preserved here via KeepAliveConfig:

  • knob < 0 => keepalive disabled entirely (KeepAliveConfig.Enable = false)
  • knob == 0 => SO_KEEPALIVE on, OS defaults for idle/interval/count
  • knob > 0 => SO_KEEPALIVE on, idle = knob, interval = 5s, count = 5
View Source 
const (
	// Header sizes from HTCondor reli_sock.cpp
	NormalHeaderSize = 5
	MaxHeaderSize    = NormalHeaderSize // TODO: Add MAC_SIZE when implementing MD

	// Maximum message size from HTCondor (1MB)
	MaxMessageSize = 1024 * 1024

	// Frame size threshold - send frame when message reaches this size
	DefaultFrameThreshold = 4096 // 4KB default threshold

	// End flag values
	EndFlagPartial  = 0 // More frames follow
	EndFlagComplete = 1 // Last frame in message
)

CEDAR protocol constants based on HTCondor's reli_sock.cpp

Variables

This section is empty.

Functions

This section is empty.

Types

type KeepAliveConfig added in v0.5.0 

type KeepAliveConfig struct {
	// Enable turns SO_KEEPALIVE on (true) or off (false).
	Enable bool
	// Idle is the time a connection is idle before the first keepalive probe
	// is sent (TCP_KEEPIDLE / TCP_KEEPALIVE). A value <= 0 leaves the OS
	// default in place.
	Idle time.Duration
	// Interval is the time between successive keepalive probes
	// (TCP_KEEPINTVL). A value <= 0 leaves the OS default in place.
	Interval time.Duration
	// Count is the number of unacknowledged probes before the connection is
	// considered dead (TCP_KEEPCNT). A value <= 0 leaves the OS default.
	Count int
}

KeepAliveConfig controls TCP keepalive probing on a CEDAR socket. The zero value keeps keepalives disabled; use DefaultKeepAliveConfig for the HTCondor-matching defaults, then override individual fields as needed.

func DefaultKeepAliveConfig added in v0.5.0 

func DefaultKeepAliveConfig() KeepAliveConfig

DefaultKeepAliveConfig returns the CEDAR keepalive settings that mirror C++ HTCondor's defaults (SO_KEEPALIVE on; idle 360s, interval 5s, count 5).

func (KeepAliveConfig) Apply added in v0.5.0 

func (k KeepAliveConfig) Apply(conn net.Conn) error

Apply configures TCP keepalives on conn according to k. It is a no-op that returns nil for connections that are not *net.TCPConn (e.g. Unix sockets used by the shared-port local path, or test pipes), so callers can apply it unconditionally. Fields set to <= 0 are translated to net.KeepAliveConfig's "leave unchanged" sentinel (-1) so the OS default is preserved for that knob.

type Stream 

type Stream struct {
	// contains filtered or unexported fields
}

Stream represents a CEDAR protocol stream over a TCP connection

func NewStream 

func NewStream(conn net.Conn) *Stream

NewStream creates a new CEDAR stream from a TCP connection

func NewStreamWithCryptoState added in v0.5.1 

func NewStreamWithCryptoState(conn net.Conn, blob []byte) (*Stream, error)

NewStreamWithCryptoState builds a Stream around conn (as NewStream does) and then restores the live crypto+framing state previously produced by ExportCryptoState, so the returned stream continues the same AES-256-GCM session byte-exactly. The AEAD is rebuilt from the exported key exactly as SetSymmetricKey constructs it, but WITHOUT regenerating the IV or zeroing the counters -- the exported IVs/counters/digests/flags are restored verbatim.

It rejects a version mismatch, a truncated blob, or a bad magic so a stale or corrupt blob fails loudly. conn is typically the SCM_RIGHTS-passed fd wrapped as a net.Conn; the blob describes the session that fd is carrying.

func (*Stream) Close 

func (s *Stream) Close() error

Close closes the underlying connection

func (*Stream) EndMessage 

func (s *Stream) EndMessage(ctx context.Context) error

EndMessage indicates end of message and sends any remaining buffered data

func (*Stream) EndMessageRead 

func (s *Stream) EndMessageRead() error

EndMessageRead indicates that message reading is complete Returns error if not all bytes have been consumed

func (*Stream) ExportCryptoState added in v0.5.1 

func (s *Stream) ExportCryptoState() ([]byte, error)

ExportCryptoState serializes the full live AES-256-GCM crypto and framing state required to resume this session byte-exactly on a different connection.

It returns an error unless the stream is at a CLEAN MESSAGE BOUNDARY of an established, encrypted session. The clean-boundary check is the core safety guarantee: exporting mid-frame would strand buffered plaintext/ciphertext that the rebuilt stream could never account for, silently desynchronizing the GCM counter/IV stream (the "buffered-bytes hazard"). Each rejection names the offending condition so the caller fails loudly rather than corrupting data.

The exported fields and WHY each is needed for byte-exact continuation:

  • encryptKey (32B): rebuilds the AEAD on the far side. Same key -> same GCM.
  • encryptIV[16] / decryptIV[16]: the per-direction BASE IVs. The low 4 bytes hold a base counter to which the message counter is added for every frame; the base IV is transmitted inline only on the counter==0 frame. Resuming with counters >= 1 means NO IV is re-sent, so both sides must already agree on these base IVs or every subsequent GCM nonce diverges.
  • encryptCounter / decryptCounter: the per-direction message counters (>= 1 on an established session). They select the GCM nonce for the next frame; off-by-one here is an immediate auth-tag failure.
  • finishedSendAAD / finishedRecvAAD: mark that each direction's first encrypted frame (which carries the 64-byte handshake-digest AAD) is already past. Restored as true so the rebuilt stream uses header-only AAD for all further frames -- matching the peer, which is likewise past its first frame.
  • finalSendDigest / finalRecvDigest: the frozen 32-byte handshake digests that formed the first-frame AAD. Not consulted again once the finished flags are true, but exported for completeness and forward compatibility.
  • sendDigestWritten / recvDigestWritten: whether any pre-encryption bytes fed the handshake digests; preserved so a re-finalization would reproduce the same digest. Inert post-handshake but kept for a faithful restore.
  • encrypted / authenticated flags and peerAddr: session status + identity.

The blob is a versioned, self-describing binary layout (magic + version) so a future field addition is detectable rather than silently misread.

func (*Stream) GetConnection 

func (s *Stream) GetConnection() net.Conn

GetConnection returns the underlying connection for TLS upgrade

func (*Stream) GetEncryption 

func (s *Stream) GetEncryption() bool

GetEncryption returns true if encryption is currently enabled Based on HTCondor's Stream::get_encryption() from stream.cpp

func (*Stream) GetFile 

func (s *Stream) GetFile(ctx context.Context, filename string) (int64, error)

GetFile receives a file from the stream Based on HTCondor's ReliSock::get_file() from reli_sock.cpp Returns the number of bytes received

func (*Stream) GetPeerAddr 

func (s *Stream) GetPeerAddr() string

GetPeerAddr returns the remote address of the connection in HTCondor sinful string format

func (*Stream) GetSecret 

func (s *Stream) GetSecret(ctx context.Context) (string, error)

GetSecret receives a secret string with automatic encryption Based on HTCondor's Stream::get_secret() from stream.cpp Temporarily enables encryption if possible, then restores previous state

func (*Stream) GetTimeout 

func (s *Stream) GetTimeout() time.Duration

GetTimeout returns the current socket timeout duration

func (*Stream) IsAuthenticated 

func (s *Stream) IsAuthenticated() bool

IsAuthenticated returns true if the stream has completed authentication

func (*Stream) IsConnected 

func (s *Stream) IsConnected() bool

IsConnected returns true if the underlying connection is still open

func (*Stream) IsEncrypted 

func (s *Stream) IsEncrypted() bool

IsEncrypted returns true if the stream is using encryption

func (*Stream) PutFile 

func (s *Stream) PutFile(ctx context.Context, filename string) (int64, error)

PutFile sends a file over the stream Based on HTCondor's ReliSock::put_file() from reli_sock.cpp Returns the number of bytes sent

func (*Stream) PutSecret 

func (s *Stream) PutSecret(ctx context.Context, secret string) error

PutSecret sends a secret string with automatic encryption Based on HTCondor's Stream::put_secret() from stream.cpp Temporarily enables encryption if possible, then restores previous state

func (*Stream) ReadFrame 

func (s *Stream) ReadFrame(ctx context.Context) ([]byte, bool, error)

ReadFrame reads a single frame from the stream and returns the data and EOM flag

func (*Stream) ReadMessageBytes 

func (s *Stream) ReadMessageBytes(ctx context.Context, data []byte) (int, error)

ReadMessageBytes reads up to n bytes from the current message Returns error if trying to read more bytes than available in current message

func (*Stream) ReceiveCompleteMessage 

func (s *Stream) ReceiveCompleteMessage(ctx context.Context) ([]byte, error)

ReceiveCompleteMessage receives a complete message, reading multiple frames if necessary This is the main method for reading complete messages that may span multiple frames

func (*Stream) ReceiveFrame 

func (s *Stream) ReceiveFrame(ctx context.Context) ([]byte, error)

ReceiveFrame receives and deframes a message from the stream Uses HTCondor CEDAR protocol format: [1 byte: end flag] [4 bytes: message length in network order] [message data]

func (*Stream) ReceiveFrameWithEnd 

func (s *Stream) ReceiveFrameWithEnd(ctx context.Context) ([]byte, byte, error)

ReceiveFrameWithEnd receives a message and returns both data and end flag

func (*Stream) SendMessage 

func (s *Stream) SendMessage(ctx context.Context, data []byte) error

SendMessage sends a framed message over the stream Uses HTCondor CEDAR protocol format: [1 byte: end flag] [4 bytes: message length in network order] [message data]

func (*Stream) SendPartialMessage 

func (s *Stream) SendPartialMessage(ctx context.Context, data []byte) error

SendPartialMessage sends a message frame (end flag = 0)

func (*Stream) SetAuthenticated 

func (s *Stream) SetAuthenticated(authenticated bool)

SetAuthenticated sets the authentication status of the stream

func (*Stream) SetConnection 

func (s *Stream) SetConnection(conn net.Conn)

SetConnection replaces the underlying connection (e.g., with TLS connection)

func (*Stream) SetCryptoMode 

func (s *Stream) SetCryptoMode(enabled bool) bool

SetCryptoMode enables or disables encryption on the stream Based on HTCondor's Stream::set_crypto_mode() from stream.cpp Returns false if encryption cannot be enabled (e.g., no key exchanged)

func (*Stream) SetEncrypted 

func (s *Stream) SetEncrypted(encrypted bool)

SetEncrypted sets the encryption status of the stream

func (*Stream) SetPeerAddr 

func (s *Stream) SetPeerAddr(addr string)

SetPeerAddr sets the remote address (useful when the address should be in a specific format)

func (*Stream) SetSymmetricKey 

func (s *Stream) SetSymmetricKey(key []byte) error

SetSymmetricKey configures AES-GCM encryption with the provided key

func (*Stream) SetTimeout 

func (s *Stream) SetTimeout(duration time.Duration) error

SetTimeout sets the socket timeout duration Based on HTCondor's Stream::timeout() from stream.cpp A timeout of 0 means no timeout (blocking indefinitely)

func (*Stream) StartMessage 

func (s *Stream) StartMessage()

StartMessage resets EOM state to allow writing a new message

func (*Stream) StartMessageRead 

func (s *Stream) StartMessageRead(ctx context.Context) error

StartMessageRead begins reading a complete message (potentially across multiple frames)

func (*Stream) WriteFrame 

func (s *Stream) WriteFrame(ctx context.Context, data []byte, isEOM bool) error

WriteFrame writes a single frame to the stream with the EOM flag

func (*Stream) WriteMessage 

func (s *Stream) WriteMessage(ctx context.Context, data []byte) error

WriteMessage writes data to the message buffer Data is accumulated until EndMessage() is called or frame threshold is reached

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.