mirror of
https://github.com/openziti/ziti.git
synced 2026-10-13 15:51:21 +00:00
Prevent payload which has looped back to sender from being acked
This commit is contained in:
@@ -107,8 +107,9 @@ func (buffer *LinkReceiveBuffer) Inspect() *inspect.XgressRecvBufferDetail {
|
||||
|
||||
func (buffer *LinkReceiveBuffer) inspectComplete() *inspect.XgressRecvBufferDetail {
|
||||
nextPayload := "none"
|
||||
if head := buffer.PeekHead(); head != nil {
|
||||
nextPayload = fmt.Sprintf("%v", head.Sequence)
|
||||
if head := buffer.tree.LeftValue(); head != nil {
|
||||
payload := head.(*Payload)
|
||||
nextPayload = fmt.Sprintf("%v", payload.Sequence)
|
||||
}
|
||||
|
||||
return &inspect.XgressRecvBufferDetail{
|
||||
|
||||
@@ -66,7 +66,7 @@ type PayloadFlag uint32
|
||||
|
||||
const (
|
||||
PayloadFlagCircuitEnd PayloadFlag = 1
|
||||
PayloadFlagEgress PayloadFlag = 2
|
||||
PayloadFlagOriginator PayloadFlag = 2
|
||||
PayloadFlagCircuitStart PayloadFlag = 4
|
||||
)
|
||||
|
||||
@@ -86,7 +86,7 @@ func (header *Header) GetFlags() string {
|
||||
}
|
||||
|
||||
func (header *Header) GetOriginator() Originator {
|
||||
if isPayloadFlagSet(header.Flags, PayloadFlagEgress) {
|
||||
if isPayloadFlagSet(header.Flags, PayloadFlagOriginator) {
|
||||
return Terminator
|
||||
}
|
||||
return Initiator
|
||||
@@ -267,9 +267,9 @@ func (payload *Payload) IsCircuitStartFlagSet() bool {
|
||||
|
||||
func SetOriginatorFlag(flags uint32, originator Originator) uint32 {
|
||||
if originator == Initiator {
|
||||
return ^uint32(PayloadFlagEgress) & flags
|
||||
return ^uint32(PayloadFlagOriginator) & flags
|
||||
}
|
||||
return uint32(PayloadFlagEgress) | flags
|
||||
return uint32(PayloadFlagOriginator) | flags
|
||||
}
|
||||
|
||||
func (payload *Payload) GetLoggerFields() logrus.Fields {
|
||||
|
||||
@@ -55,14 +55,14 @@ func TestSetOriginatorFlag(t *testing.T) {
|
||||
flags: 0,
|
||||
originator: Terminator,
|
||||
},
|
||||
want: uint32(PayloadFlagEgress),
|
||||
want: uint32(PayloadFlagOriginator),
|
||||
},
|
||||
{name: "set end of circuit to egress",
|
||||
args: args{
|
||||
flags: uint32(PayloadFlagCircuitEnd),
|
||||
originator: Terminator,
|
||||
},
|
||||
want: uint32(PayloadFlagCircuitEnd) | uint32(PayloadFlagEgress),
|
||||
want: uint32(PayloadFlagCircuitEnd) | uint32(PayloadFlagOriginator),
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
|
||||
@@ -564,7 +564,10 @@ func (self *Xgress) nextReceiveSequence() int32 {
|
||||
func (self *Xgress) PayloadReceived(payload *Payload) {
|
||||
log := pfxlog.ContextLogger(self.Label()).WithFields(payload.GetLoggerFields())
|
||||
log.Debug("payload received")
|
||||
if self.linkRxBuffer.ReceiveUnordered(payload, self.Options.RxBufferSize) {
|
||||
if self.originator == payload.GetOriginator() {
|
||||
// a payload sent from this xgress has arrived back at this xgress, instead of the other end
|
||||
log.Warn("ouroboros (circuit cycle) detected, dropping payload")
|
||||
} else if self.linkRxBuffer.ReceiveUnordered(payload, self.Options.RxBufferSize) {
|
||||
log.Debug("ready to acknowledge")
|
||||
|
||||
ack := NewAcknowledgement(self.circuitId, self.originator)
|
||||
|
||||
Reference in New Issue
Block a user