routeIncoming routes one incoming frame to its session, creating the session (and dialing upstream) if this is a SYN. owner is the clientID of the requesting client; non-SYN frames for an existing session are rejected when they come from a different client (collision or spoof).
(f *frame.Frame, owner [frame.ClientIDLen]byte)
| 486 | // requesting client; non-SYN frames for an existing session are rejected when |
| 487 | // they come from a different client (collision or spoof). |
| 488 | func (s *Server) routeIncoming(f *frame.Frame, owner [frame.ClientIDLen]byte) { |
| 489 | s.mu.Lock() |
| 490 | sess, exists := s.sessions[f.SessionID] |
| 491 | existingOwner, hasOwner := s.sessionOwners[f.SessionID] |
| 492 | s.mu.Unlock() |
| 493 | |
| 494 | if exists && hasOwner && existingOwner != owner { |
| 495 | // Different client claiming an active session ID — astronomically |
| 496 | // unlikely with random 16-byte IDs, but possible if a client reused an |
| 497 | // ID from a previous process. Reject to keep clients isolated. |
| 498 | log.Printf("[exit] cross-client session collision on %x; sending RST to %x", |
| 499 | f.SessionID[:4], owner[:4]) |
| 500 | s.queueRST(owner, f.SessionID) |
| 501 | s.stats.rstSent.Add(1) |
| 502 | return |
| 503 | } |
| 504 | |
| 505 | if !exists { |
| 506 | if !f.HasFlag(frame.FlagSYN) { |
| 507 | if protocol.IsProbePayload(f.Payload) { |
| 508 | s.queueVersionResponse(owner, f.SessionID) |
| 509 | return |
| 510 | } |
| 511 | log.Printf("[exit] frame for unknown session (no SYN), sending RST") |
| 512 | s.queueRST(owner, f.SessionID) |
| 513 | s.stats.rstSent.Add(1) |
| 514 | return |
| 515 | } |
| 516 | if s.isDialSuppressed(f.Target) { |
| 517 | log.Printf("[exit] dial suppressed for %s (recent failure backoff); sending RST", f.Target) |
| 518 | s.queueRST(owner, f.SessionID) |
| 519 | s.stats.rstSent.Add(1) |
| 520 | return |
| 521 | } |
| 522 | var err error |
| 523 | sess, err = s.openSession(f.SessionID, f.Target, owner) |
| 524 | if err != nil { |
| 525 | s.recordDialFailure(f.Target, err) |
| 526 | s.stats.dialsFail.Add(1) |
| 527 | log.Printf("[exit] dial %s: %v", f.Target, err) |
| 528 | return |
| 529 | } |
| 530 | s.stats.dialsOK.Add(1) |
| 531 | s.clearDialFailure(f.Target) |
| 532 | } |
| 533 | sess.ProcessRx(f) |
| 534 | // Touch activity AFTER ProcessRx so a successful client→server frame |
| 535 | // resets the idle timer for this session. |
| 536 | s.mu.Lock() |
| 537 | if _, stillExists := s.sessions[f.SessionID]; stillExists { |
| 538 | s.lastActivity[f.SessionID] = time.Now() |
| 539 | } |
| 540 | s.mu.Unlock() |
| 541 | } |
| 542 | |
| 543 | // queueRST enqueues a RST frame for the given session to be delivered to |
| 544 | // owner on its next poll. Also wakes that client's long-poll so the RST is |