Merge pull request #115 from smartcontractkit/copy-cb8e5eb65c417703b7d35c96a95b9e13a2d14020

Copy cb8e5eb65c417703b7d35c96a95b9e13a2d14020


Co-authored-by: Kostis Karantias <[email protected]>
Co-authored-by: Philipp Schindler <[email protected]>
Co-authored-by: stchrysa <[email protected]>

Author: 4 people

README.md

@@ -1,29 +1,36 @@
 # libocr
 
-libocr consists of a Go library and a set of Solidity smart contracts that implement the *Chainlink Offchain Reporting Protocol*, a [Byzantine fault tolerant](https://en.wikipedia.org/wiki/Byzantine_fault) protocol that allows a set of oracles to generate *offchain* an aggregate report of the oracles' observations of some underlying data source. This report is then transmitted to an onchain contract in a single transaction.
+libocr consists of a Go library and a set of Solidity smart contracts that implements various versions of the *Chainlink Offchain Reporting Protocol*, a [Byzantine fault tolerant](https://en.wikipedia.org/wiki/Byzantine_fault) "consensus" protocol that allows a set of oracles to generate *offchain* an aggregate report of the oracles' observations of some underlying data source. This report is then transmitted to an onchain contract in a single transaction.
 
-You may also be interested in [libocr's integration into the actual Chainlink node](https://github.com/smartcontractkit/chainlink/tree/develop/core/services/offchainreporting).
+You may also be interested in libocr's integration into the actual Chainlink node. ([V1](https://github.com/smartcontractkit/chainlink/tree/develop/core/services/ocr) [V2](https://github.com/smartcontractkit/chainlink/tree/develop/core/services/ocr2) [V3](https://github.com/smartcontractkit/chainlink/tree/develop/core/services/ocr3))
 
 
 ## Protocol Description
 
-Protocol execution mostly happens offchain over a peer to peer network between Chainlink nodes. The nodes regularly elect a new leader node who drives the rest of the protocol. The protocol is designed to choose each leader fairly and quickly rotate away from leaders that aren’t making progress towards timely onchain reports.
+Please see the whitepapers available at https://chainlinklabs.com/research for detailed protocol descriptions.
 
-The leader regularly requests followers to provide freshly signed observations and aggregates them into a report. It then sends the aggregate report back to the followers and asks them to attest to the report's validity by signing it. If a quorum of followers approves the report, the leader assembles a final report with the quorum's signatures and broadcasts it to all followers.
-
-The nodes then attempt to transmit the final report to the smart contract according to a randomized schedule. Finally, the smart contract verifies that a quorum of nodes signed the report and exposes the median value to consumers.
+## Protocol Versions
 
+- OCR1 is deprecated and being phased out.
+- OCR2 & OCR3 are in production.
+- OCR3.1 is in alpha and excluded from any bug bounties at this time.
 
 ## Organization
 ```
-.
-├── contract: Ethereum smart contracts
+├── bigbigendian: helper package
+├── commontypes: shared type definitions
+├── contract: OCR1 Ethereum contracts
+├── contract2: OCR2 Ethereum contracts
+├── contract3: OCR3 Ethereum contracts
 ├── gethwrappers: go-ethereum bindings for the OCR1 contracts, generated with abigen
 ├── gethwrappers2: go-ethereum bindings for the OCR2 contracts, generated with abigen
-├── networking: p2p networking layer
-├── offchainreporting: offchain reporting protocol version 1
-├── offchainreporting2: offchain reporting protocol version 2 specific packages, not much here
-├── offchainreporting2plus: offchain reporting protocol version 2 and beyond
-├── permutation: helper package for generating permutations
-└── subprocesses: helper package for managing go routines
+├── gethwrappers3: go-ethereum bindings for the OCR3 contracts, generated with abigen
+├── networking: OCR networking layer
+├── offchainreporting: OCR1
+├── offchainreporting2: OCR2-specific
+├── offchainreporting2plus: OCR2 and beyond (These versions share many interface definitions to make integration of new versions easier)
+├── permutation: helper package
+├── quorumhelper: helper package
+├── ragep2p: p2p networking
+└── subprocesses: helper package
 ```

commontypes/types.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"strings"
 
+	// TODO: is there a way to remove this dependency?
 	ragetypes "github.com/smartcontractkit/libocr/ragep2p/types"
 )
 

go.mod

@@ -1,8 +1,8 @@
 module github.com/smartcontractkit/libocr
 
-go 1.23.0
+go 1.24
 
-toolchain go1.23.6
+toolchain go1.24.4
 
 require (
 	github.com/ethereum/go-ethereum v1.15.3
@@ -65,7 +65,7 @@ require (
 	github.com/influxdata/influxdb1-client v0.0.0-20220302092344-a9ab5670611c // indirect
 	github.com/influxdata/line-protocol v0.0.0-20210311194329-9aa0e372d097 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
-	github.com/klauspost/compress v1.16.0 // indirect
+	github.com/klauspost/compress v1.17.11 // indirect
 	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
@@ -102,7 +102,7 @@ require (
 	github.com/urfave/cli/v2 v2.27.5 // indirect
 	github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 // indirect
 	golang.org/x/exp v0.0.0-20240416160154-fe59bbe5cc7f // indirect
-	golang.org/x/net v0.36.0 // indirect
+	golang.org/x/net v0.34.0 // indirect
 	golang.org/x/sync v0.11.0 // indirect
 	golang.org/x/sys v0.30.0 // indirect
 	golang.org/x/text v0.22.0 // indirect

go.sum

@@ -141,8 +141,8 @@ github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7Bd
 github.com/jackpal/go-nat-pmp v1.0.2/go.mod h1:QPH045xvCAeXUZOxsnwmrtiCoxIr9eob+4orBN1SBKc=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/klauspost/compress v1.16.0 h1:iULayQNOReoYUe+1qtKOqw9CwJv3aNQu8ivo7lw1HU4=
-github.com/klauspost/compress v1.16.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
+github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
 github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
@@ -312,8 +312,8 @@ golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
-golang.org/x/net v0.36.0 h1:vWF2fRbw4qslQsQzgFqZff+BItCvGFQqKzKIzx1rmoA=
-golang.org/x/net v0.36.0/go.mod h1:bFmbeoIPfrw4sMHNhb4J9f6+tPziuGjq7Jk/38fxi1I=
+golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
+golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=

internal/ringbuffer/ringbuffer.go

@@ -0,0 +1,93 @@
+package ringbuffer
+
+import "fmt"
+
+// RingBuffer implements a fixed capacity ring buffer for items of type T.
+// NOTE: THIS IMPLEMENTATION IS NOT SAFE FOR CONCURRENT USE.
+type RingBuffer[T any] struct {
+	first int // index of the front (=oldest) element
+	size  int // number of elements currently stored in this ring buffer
+	items []T // fixed size buffer holding the elements
+}
+
+func NewRingBuffer[T any](cap int) *RingBuffer[T] {
+	if cap <= 0 {
+		panic(fmt.Sprintf("NewRingBuffer: cap must be positive, got %d", cap))
+	}
+	return &RingBuffer[T]{
+		0,
+		0,
+		make([]T, cap),
+	}
+}
+
+func (rb *RingBuffer[T]) Size() int {
+	return rb.size
+}
+
+func (rb *RingBuffer[T]) Cap() int {
+	return len(rb.items)
+}
+
+func (rb *RingBuffer[T]) IsEmpty() bool {
+	return rb.size == 0
+}
+
+func (rb *RingBuffer[T]) IsFull() bool {
+	return rb.size == len(rb.items)
+}
+
+// Peek returns the front (=oldest) item without removing it.
+// Return false as second argument if there are no items in the ring buffer.
+func (rb *RingBuffer[T]) Peek() (result T, ok bool) {
+	if rb.size > 0 {
+		ok = true
+		result = rb.items[rb.first]
+	}
+	return result, ok
+}
+
+// Pop removes and returns the front (=oldest) item.
+// Return false as second argument if there are no items in the ring buffer.
+func (rb *RingBuffer[T]) Pop() (result T, ok bool) {
+	result, ok = rb.Peek()
+	if ok {
+		var zero T
+		rb.items[rb.first] = zero
+		rb.first = (rb.first + 1) % len(rb.items)
+		rb.size--
+	}
+	return result, ok
+}
+
+// Try to push a new item to the back of the ring buffer.
+// Returns
+//   - true if the item was added, or
+//   - false if the item cannot be added because the buffer is currently full.
+func (rb *RingBuffer[T]) TryPush(item T) (ok bool) {
+	if rb.IsFull() {
+		return false
+	}
+	rb.items[(rb.first+rb.size)%len(rb.items)] = item
+	rb.size++
+	return true
+}
+
+// Push new item to the back of the ring buffer.
+// If the buffer is currently full, the front (=oldest) item is evicted and returned to make space for the new item.
+func (rb *RingBuffer[T]) PushEvict(item T) (evicted T, didEvict bool) {
+	if rb.IsFull() {
+		// Evict the oldest item to be returned.
+		evicted = rb.items[rb.first]
+		didEvict = true
+
+		// Push the new item to new empty space and update the first index to the next (oldest) item.
+		rb.items[rb.first] = item
+		rb.first = (rb.first + 1) % len(rb.items)
+	} else {
+		// Perform a normal push operation (which is known to be successful as the buffer is not full).
+		rb.items[(rb.first+rb.size)%len(rb.items)] = item
+		rb.size++
+	}
+	return evicted, didEvict
+}

internal/util/generic.go

@@ -11,3 +11,11 @@ func NilCoalesce[T any](maybe *T, default_ T) T {
 		return default_
 	}
 }
+
+func NilCoalesceSlice[T any](maybe []T) []T {
+	if maybe != nil {
+		return maybe
+	} else {
+		return []T{}
+	}
+}

networking/internal/ocrendpointv3/responselimit/checker.go

@@ -0,0 +1,180 @@
+package responselimit
+
+import (
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/smartcontractkit/libocr/networking/internal/ocrendpointv3/types"
+)
+
+type ResponseCheckResult byte
+
+// Enum specifying the list of return values for responseChecker.CheckResponse(...).
+const (
+	// A response is rejected if the policy
+	// (1) was not found, or
+	// (2) was expired, or
+	// (3) was found but decided to reject the request.
+	//
+	// As policies are automatically cleaned up (in some non-deterministic manner), there is no way to distinguish
+	// cases (1) and (2), and for simplicity also case (3) is handled identically.
+	//
+	// We intentionally use 0 as the first enum value for Reject as a safe default here.
+	ResponseCheckResultReject ResponseCheckResult = iota
+
+	// A (non-expired) policy was found, and the policy did decide that the response should be allowed.
+	ResponseCheckResultAllow
+)
+
+type responseCheckerMapEntry struct {
+	index    int
+	policy   ResponsePolicy
+	streamID types.StreamID
+}
+
+// Data structure for keeping track of open requests until a set expiry date.
+//
+// Cleanup of expired entries is performed automatically. Whenever a new entry is added, two random entries are checked
+// and removed if expired. This ensures that, on expectation, the number of tracked entries is approx. 2x the number
+// of non-expired entries.
+//
+// SetPolicy(...) and CheckResponse(...) are O(1) operations.
+type ResponseChecker struct {
+	mutex    sync.Mutex
+	rids     []types.RequestID
+	policies map[types.RequestID]responseCheckerMapEntry
+	rng      *rand.Rand
+}
+
+func NewResponseChecker() *ResponseChecker {
+	return &ResponseChecker{
+		sync.Mutex{},
+		make([]types.RequestID, 0),
+		make(map[types.RequestID]responseCheckerMapEntry),
+		rand.New(rand.NewSource(time.Now().UnixNano())),
+	}
+}
+
+// Sets the policy for a given (fresh) request ID. After setting the policy, calling Pop(...) for the same ID before the
+// policy expires returns the policy Set with this function. If a policy with the provided ID is already present, it
+// will be overwritten.
+func (c *ResponseChecker) SetPolicy(sid types.StreamID, rid types.RequestID, policy ResponsePolicy) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	// Lookup an existing policy for the provided request ID.
+	// If it exists, we override the policy, keeping its location at the prior index.
+	// Otherwise, we need use a new index and also track the request ID in the c.rids list.
+	entry, exists := c.policies[rid]
+	if exists {
+		entry = responseCheckerMapEntry{entry.index, policy, sid}
+	} else {
+		// We set entry.index = len(c.rids) to let it point to the request ID we will append to c.rids list.
+		entry = responseCheckerMapEntry{len(c.rids), policy, sid}
+		c.rids = append(c.rids, rid)
+	}
+
+	// Actually save the policy update back to the c.policies map.
+	c.policies[rid] = entry
+
+	// If the number of tracked policies increased, we check 2 random policies and remove them if expired. This way
+	// the number of tracked policies only grows to 2x the number of non-expired policies in expectation.
+	if !exists {
+		c.cleanupExpired()
+	}
+}
+
+// Lookup the policy for a given response and check if it should be allowed or rejected.
+// See responseCheckResult for additional documentation on the potential return values of this function.
+func (c *ResponseChecker) CheckResponse(sid types.StreamID, rid types.RequestID, size int) ResponseCheckResult {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	entry, exists := c.policies[rid]
+	if !exists {
+		return ResponseCheckResultReject
+	}
+	if entry.streamID != sid {
+		return ResponseCheckResultReject
+	}
+
+	now := time.Now()
+	if entry.policy.isPolicyExpired(now) {
+		c.removeEntry(rid, entry.index)
+		return ResponseCheckResultReject
+	}
+
+	policyResult := entry.policy.checkResponse(rid, size, now)
+
+	// Recheck the policy of expiry, useful to cleanup one-time-use policies immediately.
+	if entry.policy.isPolicyExpired(now) {
+		c.removeEntry(rid, entry.index)
+	}
+
+	return policyResult
+}
+
+// Removes all currently tracked policies for the given stream ID. To ensure that responses sent to a stream cannot be
+// accepted after this stream is closed and reopened, this function is called when the Stream is closed (and removed
+// from the demuxer).
+func (c *ResponseChecker) ClearPoliciesForStream(sid types.StreamID) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	for i := 0; i < len(c.rids); i++ {
+		rid := c.rids[i]
+		policy := c.policies[rid]
+
+		if policy.streamID == sid {
+			// We found a policy which matches the given stream ID.
+			// So we remove the entry from the list of request IDs and policies.
+			c.removeEntry(rid, i)
+
+			// The above removeEntry(...) removes c.rids[i], thus in the next iteration index its value is replaced
+			// by a different request ID. We decrement index i to ensure that we don't skip the new value at index i.
+			i--
+		}
+	}
+}
+
+// Check two random policies. A checked policy is removed if it is found to be expired.
+func (c *ResponseChecker) cleanupExpired() {
+	now := time.Now()
+
+	// At most 2 iterations, enter loop body only if c.rids is non empty.
+	for i := 0; i < 2 && len(c.rids) > 0; i++ {
+		// Select a random policy.
+		index := c.rng.Intn(len(c.rids))
+		id := c.rids[index]
+		policy := c.policies[id].policy
+
+		// Remove it if it is expired.
+		if policy.isPolicyExpired(now) {
+			c.removeEntry(id, index)
+		}
+	}
+}
+
+// Remove the policy for a given request ID from (1) the map of policies and (2) the list of request IDs.
+func (c *ResponseChecker) removeEntry(id types.RequestID, index int) {
+	// Remove the entry from the map of polices.
+	delete(c.policies, id)
+
+	// Handle the "index == last-index" corner case separately.
+	// This avoids wrongfully reinserting the deleted policy.
+	if index == len(c.rids)-1 {
+		c.rids = c.rids[0 : len(c.rids)-1]
+		return
+	}
+
+	// Swap the last entry's id to the position of the to be removed id, and remove the last value from the rids list.
+	lastID := c.rids[len(c.rids)-1]
+	c.rids[index] = lastID
+	c.rids = c.rids[0 : len(c.rids)-1]
+
+	// Update the index point for the c.policies[lastId] to point to the now changed position.
+	lastEntry := c.policies[lastID]
+	lastEntry.index = index
+	c.policies[lastID] = lastEntry
+}