From 41eedec524521c770d4894970aad1433344016be Mon Sep 17 00:00:00 2001 From: Eric_Lee Date: Tue, 3 Nov 2026 18:35:31 +0800 Subject: [PATCH 1/3] fix: match login events at stream boundaries Scan available buffered chunks instead of checking every byte prefix. Defer narrative failures and expiry messages until line completion, preserve post-prompt data, bound partial matching, and add stream regressions and fuzz coverage. --- client.go | 16 ++- docs/login-stream.md | 82 ++++++++++++ login.go | 277 +++++++++++++++-------------------------- login_stream_test.go | 289 +++++++++++++++++++++++++++++++++++++++++++ login_text.go | 149 ++++++++++++++++++++++ 5 files changed, 633 insertions(+), 180 deletions(-) create mode 100644 docs/login-stream.md create mode 100644 login_stream_test.go create mode 100644 login_text.go diff --git a/client.go b/client.go index 7e0a2b7..cec5883 100644 --- a/client.go +++ b/client.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "net" + "strings" "sync" ) @@ -49,6 +50,8 @@ func NewClientConn(conn net.Conn, config *ClientConfig) (*Client, error) { func newClientConn(conn net.Conn, config clientConfig) (*Client, error) { transport := newTransport(conn, config.transportConfig()) + autoLogin := config.User != "" || config.Password != "" + transport.setTraceRedaction(autoLogin) var reader io.Reader = transport var writer io.Writer = transport if config.WrapStream != nil { @@ -62,21 +65,24 @@ func newClientConn(conn net.Conn, config clientConfig) (*Client, error) { // Keep the same buffered reader for login and the interactive session. Login // reads only through the matched prompt, so bytes that arrived in the same // transport read after that prompt remain available to Client.Read. - bufferedReader := bufio.NewReaderSize(reader, 4096) + bufferedReader := bufio.NewReaderSize(reader, loginReadBufferSize) client := &Client{ conn: conn, transport: transport, reader: bufferedReader, writer: writer, } - if config.User != "" || config.Password != "" { - transport.setTraceRedaction(true) - err := login(conn, bufferedReader, writer, config, transport.trace) - transport.setTraceRedaction(false) + if autoLogin { + replay, err := login(conn, bufferedReader, writer, config, transport.trace) if err != nil { + // Keep redaction enabled while a stream wrapper flushes on Close. _ = client.Close() return nil, fmt.Errorf("telnet: handshake failed: %w", err) } + if replay != "" { + client.reader = io.MultiReader(strings.NewReader(replay), bufferedReader) + } + transport.setTraceRedaction(false) } return client, nil } diff --git a/docs/login-stream.md b/docs/login-stream.md new file mode 100644 index 0000000..69f4b66 --- /dev/null +++ b/docs/login-stream.md @@ -0,0 +1,82 @@ +# Login stream boundaries + +`Dial` and `NewClientConn` keep the existing public API. Login and the session +share a 4 KiB buffered application reader, after Telnet framing and any +`WrapStream` charset conversion. + +## Reads are not messages + +The login loop fills the buffer only when it is empty, then scans the bytes +already available using `Peek` and `Discard`. It does not call `ReadByte` in the +login loop, wait for a full 4 KiB block, or add sleeps. A short `Password:` prompt +can therefore receive a response immediately. + +Text normalization still examines bytes to carry ANSI, UTF-8, CR/LF, and +backspace state across chunks. It does not copy the entire current line or run +all regular expressions after each byte. + +Completed CRLF/LF lines are evaluated once. At the end of an available chunk, +only interactive prompt matching is allowed, and only for a candidate of at +most 512 bytes. Failure messages and password-expiry narratives wait for a line +ending, EOF, or the overall login timeout. For example: + +```text +Authentication failed count: 0 +Password expired policy: disabled +New password policy: 90 days +``` + +These remain banners even when a read stops immediately after `failed`, +`expired`, or `New password`. An unterminated explicit failure can be reported +at EOF or timeout; finalization never sends credentials or declares success on +a closed connection. + +An unfinished password-change input must end in `:` or `：` before being +classified. `New password:` is rejected without submitting the managed old +password. Custom `PasswordChangePrompt` and `FailurePrompt` still replace their +respective defaults; their narrative matches obey the same completion rules. +Other custom prompts remain full matches against the current normalized line. + +Overlong logical lines are ignored until a line reset or ending. They are not +converted into a sliding suffix that could accidentally look like a prompt. + +## Preserving session data + +Only the consumed portion of a chunk is discarded. Bytes after the accepted +shell prompt remain in the same reader. When CR/LF was consumed to establish a +line boundary, that one- or two-byte terminator is replayed before the buffered +session data. There is no second unbounded transcript or replay buffer. + +For `router#\r\nALARM`, the next `Client.Read` still sees `\r\nALARM`; the accepted +prompt is not replayed. ANSI formatting already consumed with the prompt is +not reconstructed. The interactive stream itself is not sanitized. + +A delimiter-less textual prompt can always be the prefix of a longer message. +No TCP buffer size or fixed delay can remove that ambiguity. The defaults use +bounded prompt shapes and known-banner filtering; unusual assets should use +precise platform-specific prompt expressions. This is not a protocol-level +proof of authentication. + +## Diagnostics and malformed subnegotiations + +Trace redaction is applied centrally to wire/application bytes and to both +received and generated `SB` payloads. Option names, lengths, and login-state +events remain visible. Redaction starts before `WrapStream` is invoked and +stays enabled through failed-login cleanup, including writer flushes. After a +successful login, normal trace behavior resumes; post-login traffic and the +standalone monitor can still contain secrets. + +An oversized `SB` reports `ErrSubnegotiationTooLong` and discards its payload +until the real `IAC SE`, including across reads. Escaped `IAC IAC` does not end +discarding. Following application bytes and valid negotiations in the same +chunk are retained. A caller that continues reading after the error never +receives the oversized subnegotiation's tail as terminal data. + +## Regression checks + +The added tests exercise deterministic one-byte, split-point, 1 KiB, and 4 KiB +reads, interactive `net.Pipe` prompts without newlines, EOF/timeout results, +UTF-8 and ANSI fragments, long banners, environment subnegotiation logs, and +cleanup writes. Fuzz targets check diagnostic-banner chunking and oversized +subnegotiation recovery. `BenchmarkLoginLongBanner` measures the 1 KiB, 4 KiB, +and 64 KiB banner paths without adding timing assertions to unit tests. diff --git a/login.go b/login.go index 779fbb2..07fc03a 100644 --- a/login.go +++ b/login.go @@ -3,6 +3,7 @@ package tclientlib import ( "bufio" "bytes" + "errors" "fmt" "io" "net" @@ -10,12 +11,15 @@ import ( "strings" "sync" "time" + "unicode" "unicode/utf8" ) const ( - maxContinueResponses = 3 + maxContinueResponses = 3 maxDefaultPromptBytes = 128 + loginReadBufferSize = 4096 + maxPartialPromptBytes = 512 ) var ( @@ -156,20 +160,48 @@ const ( loginPasswordChangeRequired ) +// Narrative results are only authoritative on a completed logical line. A +// Read boundary is not an end-of-line marker. +func (m *loginMachine) messageResult(line []byte, trace *tracer) loginResult { + if len(bytes.TrimSpace(line)) == 0 { + return loginPending + } + if fullLineMatch(m.patterns.passwordChange, line) { + trace.printf("login password-change message matched phase=%s", m.phase) + return loginPasswordChangeRequired + } + if fullLineMatch(m.patterns.failure, line) { + trace.printf("login failure message matched phase=%s", m.phase) + return loginFailed + } + return loginPending +} + func (m *loginMachine) handleLine(writer io.Writer, line []byte, trace *tracer) (loginResult, error) { + if result := m.messageResult(line, trace); result != loginPending { + return result, nil + } + return m.handlePrompt(writer, line, trace) +} + +// Interactive prompts normally have no newline. Check them only at the end of +// the currently available chunk, never after every character of a banner. +func (m *loginMachine) handlePrompt(writer io.Writer, line []byte, trace *tracer) (loginResult, error) { + if len(line) > maxPartialPromptBytes || !utf8.Valid(line) { + return loginPending, nil + } line = bytes.TrimSpace(line) if len(line) == 0 { return loginPending, nil } - if fullLineMatch(m.patterns.passwordChange, line) { - trace.printf("login password-change prompt matched phase=%s", m.phase) + // Require input punctuation for an incomplete password-change prompt. + // "New password policy: ..." must not match the prefix "New password". + if (bytes.HasSuffix(line, []byte(":")) || bytes.HasSuffix(line, []byte("："))) && + fullLineMatch(m.patterns.passwordChange, line) { + trace.printf("login password-change input prompt matched phase=%s", m.phase) return loginPasswordChangeRequired, nil } - if fullLineMatch(m.patterns.failure, line) { - trace.printf("login failure prompt matched phase=%s", m.phase) - return loginFailed, nil - } if fullLineMatch(m.patterns.continuePrompt, line) { if m.continueResponses >= maxContinueResponses { trace.printf("login continue prompt ignored after limit phase=%s", m.phase) @@ -229,10 +261,13 @@ func defaultShellPromptMatches(line []byte) bool { if len(line) == 0 || len(line) > maxDefaultPromptBytes { return false } + if isBannerLikeShellPrompt(string(line)) { + return false + } if defaultMarkedBracketShellPrompt.Match(line) { return true } - if defaultTokenShellPrompt.Match(line) && !isBannerLikeShellPrompt(string(line)) { + if defaultTokenShellPrompt.Match(line) { return true } @@ -259,11 +294,12 @@ func isBannerLabel(value string) bool { func isBannerLikeShellPrompt(value string) bool { value = strings.TrimSpace(strings.TrimRight(value, "#$>%")) fields := strings.FieldsFunc(strings.ToLower(value), func(r rune) bool { - return !(r >= 'a' && r <= 'z') && !(r >= '0' && r <= '9') + return !unicode.IsLetter(r) && !unicode.IsNumber(r) }) for _, field := range fields { switch field { - case "login", "successful", "failed", "failure", "password", "warning", "notice", "error", "press", "hit", "unauthorized", "denied", "attempt", "attempts", "continue": + case "login", "successful", "failed", "failure", "password", "warning", "notice", "error", "press", "hit", "unauthorized", "denied", "attempt", "attempts", "continue", + "登录成功", "登录失败", "认证成功", "认证失败", "密码错误", "密码过期", "系统警告", "警告", "错误": return true } } @@ -279,52 +315,76 @@ func fullLineMatch(pattern *regexp.Regexp, line []byte) bool { return match != nil && match[0] == 0 && match[1] == len(line) } -func login(conn net.Conn, reader *bufio.Reader, writer io.Writer, config clientConfig, trace *tracer) error { +// login returns only a line terminator already consumed while confirming a +// prompt. All other bytes after the match stay in the shared buffered reader. +func login(conn net.Conn, reader *bufio.Reader, writer io.Writer, config clientConfig, trace *tracer) (replay string, err error) { machine := newLoginMachine(config) text := newLoginText(config.maxLoginBuffer) trace.printf("login start user=%t password=%t timeout=%s", config.User != "", config.Password != "", config.Timeout) guard := startLoginGuard(conn, config.Timeout) - defer guard.stop() + defer func() { + guard.stop() + // Do not return a usable Client if the timer won the completion race. + if guard.didExpire() && err == nil { + replay = "" + err = fmt.Errorf("%w after %s", ErrLoginTimeout, config.Timeout) + } + }() for { - value, readErr := reader.ReadByte() - if readErr == nil { - lines, current := text.feed([]byte{value}) - for _, line := range lines { - result, err := machine.handleLine(writer, line, trace) - if err != nil { - trace.printf("login write error: %v", err) - return err - } - if done, resultErr := resolveLoginResult(result, trace); done { - return resultErr + // Peek(1) fills the 4 KiB buffer without waiting for a full block. Scan + // what is already available, then discard only the bytes we consumed. + if _, readErr := reader.Peek(1); readErr != nil { + // A peer may close (or stop responding) after an error without a + // newline. Classify that final message, but never send credentials + // or infer success on a closed/timed-out connection. + if errors.Is(readErr, io.EOF) || guard.didExpire() { + if done, resultErr := resolveLoginResult(machine.messageResult(text.current(), trace), trace); done { + return "", resultErr } } - - if len(bytes.TrimSpace(current)) > 0 { - result, err := machine.handleLine(writer, current, trace) - if err != nil { - trace.printf("login write error: %v", err) - return err - } - if isLoginAction(result) { - text.consumeCurrent() - } - if done, resultErr := resolveLoginResult(result, trace); done { - return resultErr - } + if guard.didExpire() { + return "", fmt.Errorf("%w after %s", ErrLoginTimeout, config.Timeout) } - continue + return "", readErr + } + chunk, _ := reader.Peek(reader.Buffered()) + consumed, result, ending, scanErr := machine.consumeChunk(writer, text, chunk, trace) + _, _ = reader.Discard(consumed) + if scanErr != nil { + if guard.didExpire() { + return "", fmt.Errorf("%w after %s", ErrLoginTimeout, config.Timeout) + } + return "", scanErr + } + if done, resultErr := resolveLoginResult(result, trace); done { + return ending, resultErr } + } +} - if guard.didExpire() { - trace.printf("login timeout after %s", config.Timeout) - return fmt.Errorf("%w after %s", ErrLoginTimeout, config.Timeout) +func (m *loginMachine) consumeChunk(writer io.Writer, text *loginText, chunk []byte, trace *tracer) (int, loginResult, string, error) { + for index, value := range chunk { + line, ending := text.feedByte(value) + if ending == "" { + continue } - trace.printf("login read error: %v", readErr) - return readErr + result, err := m.handleLine(writer, line, trace) + if err != nil || result == loginSuccess || result == loginFailed || result == loginPasswordChangeRequired { + return index + 1, result, ending, err + } + } + + result, err := m.handlePrompt(writer, text.current(), trace) + ending := "" + if result == loginSuccess && text.pendingCR { + ending = "\r" } + if isLoginAction(result) { + text.consumeCurrent() + } + return len(chunk), result, ending, err } func isLoginAction(result loginResult) bool { @@ -374,136 +434,3 @@ func writeFull(writer io.Writer, data []byte) error { } return nil } - -type loginEscapeState uint8 - -const ( - loginEscapeNone loginEscapeState = iota - loginEscapeStart - loginEscapeCSI - loginEscapeOSC - loginEscapeOSCEscape -) - -type loginText struct { - line []byte - maxLine int - pendingCR bool - escape loginEscapeState -} - -func newLoginText(maxLine int) *loginText { - return &loginText{maxLine: maxLine} -} - -func (t *loginText) feed(input []byte) ([][]byte, []byte) { - lines := make([][]byte, 0, 2) - for _, value := range input { - if t.consumeEscape(value) { - continue - } - - if t.pendingCR { - t.pendingCR = false - if value == '\n' { - lines = append(lines, t.finishLine()) - continue - } - // A bare carriage return moves the terminal cursor back to column 0. - t.line = t.line[:0] - } - - switch value { - case 0, '\a': - continue - case '\x1b': - t.escape = loginEscapeStart - case '\r': - t.pendingCR = true - case '\n': - lines = append(lines, t.finishLine()) - case '\b', 0x7f: - t.removeLastRune() - case '\t': - t.appendByte(' ') - default: - if value < 0x20 { - continue - } - t.appendByte(value) - } - } - return lines, append([]byte(nil), t.line...) -} - -func (t *loginText) consumeEscape(value byte) bool { - switch t.escape { - case loginEscapeNone: - return false - case loginEscapeStart: - switch value { - case '[': - t.escape = loginEscapeCSI - case ']': - t.escape = loginEscapeOSC - default: - t.escape = loginEscapeNone - } - return true - case loginEscapeCSI: - if value >= 0x40 && value <= 0x7e { - t.escape = loginEscapeNone - } - return true - case loginEscapeOSC: - switch value { - case '\a': - t.escape = loginEscapeNone - case '\x1b': - t.escape = loginEscapeOSCEscape - } - return true - case loginEscapeOSCEscape: - if value == '\\' { - t.escape = loginEscapeNone - } else { - t.escape = loginEscapeOSC - } - return true - default: - t.escape = loginEscapeNone - return true - } -} - -func (t *loginText) finishLine() []byte { - line := append([]byte(nil), t.line...) - t.line = t.line[:0] - return line -} - -func (t *loginText) consumeCurrent() { - t.line = t.line[:0] - t.pendingCR = false -} - -func (t *loginText) appendByte(value byte) { - t.line = append(t.line, value) - if len(t.line) <= t.maxLine { - return - } - drop := len(t.line) - t.maxLine - copy(t.line, t.line[drop:]) - t.line = t.line[:t.maxLine] -} - -func (t *loginText) removeLastRune() { - if len(t.line) == 0 { - return - } - _, size := utf8.DecodeLastRune(t.line) - if size <= 0 { - size = 1 - } - t.line = t.line[:len(t.line)-size] -} diff --git a/login_stream_test.go b/login_stream_test.go new file mode 100644 index 0000000..662546d --- /dev/null +++ b/login_stream_test.go @@ -0,0 +1,289 @@ +package tclientlib + +import ( + "bytes" + "errors" + "fmt" + "io" + "net" + "strings" + "sync/atomic" + "testing" + "time" +) + +// streamTestConn makes Read boundaries deterministic without treating them as +// message boundaries. Separate net.Pipe tests exercise interactive blocking. +type streamTestConn struct { + chunks [][]byte + writes bytes.Buffer + closed atomic.Bool + lastError error +} + +func (c *streamTestConn) Read(p []byte) (int, error) { + if c.closed.Load() { + return 0, net.ErrClosed + } + for len(c.chunks) > 0 && len(c.chunks[0]) == 0 { + c.chunks = c.chunks[1:] + } + if len(c.chunks) == 0 { + return 0, io.EOF + } + n := copy(p, c.chunks[0]) + c.chunks[0] = c.chunks[0][n:] + if len(c.chunks[0]) == 0 { + c.chunks = c.chunks[1:] + if len(c.chunks) == 0 && c.lastError != nil { + return n, c.lastError + } + } + return n, nil +} + +func (c *streamTestConn) Write(p []byte) (int, error) { + if c.closed.Load() { + return 0, net.ErrClosed + } + return c.writes.Write(p) +} +func (c *streamTestConn) Close() error { c.closed.Store(true); return nil } +func (c *streamTestConn) LocalAddr() net.Addr { return streamTestAddr("local") } +func (c *streamTestConn) RemoteAddr() net.Addr { return streamTestAddr("remote") } +func (c *streamTestConn) SetDeadline(time.Time) error { return nil } +func (c *streamTestConn) SetReadDeadline(time.Time) error { return nil } +func (c *streamTestConn) SetWriteDeadline(time.Time) error { return nil } + +type streamTestAddr string + +func (a streamTestAddr) Network() string { return "test" } +func (a streamTestAddr) String() string { return string(a) } + +func streamChunks(data []byte, size int) [][]byte { + var chunks [][]byte + for len(data) > 0 { + n := size + if n > len(data) { + n = len(data) + } + chunks = append(chunks, data[:n]) + data = data[n:] + } + return chunks +} + +func checkBannerStream(t *testing.T, parts [][]byte) { + t.Helper() + conn := &streamTestConn{chunks: parts, lastError: io.EOF} + client, err := NewClientConn(conn, &ClientConfig{User: "admin", Password: "secret", Timeout: 5 * time.Second}) + if err != nil { + t.Fatalf("login: %v", err) + } + defer client.Close() + got, err := io.ReadAll(client) + if err != nil { + t.Fatal(err) + } + if string(got) != "\r\nTAIL" { + t.Fatalf("unconsumed data = %q, want CRLF+TAIL", got) + } + if conn.writes.String() != "admin\r\nsecret\r\n" { + t.Fatalf("writes = %q", conn.writes.String()) + } +} + +func TestLoginStreamDoesNotClassifyMessagePrefixes(t *testing.T) { + for _, banner := range []string{ + "Authentication failed count: 0", "Authentication failure count: 0", + "Login failed attempts: 5", "Password expired policy: disabled", + "New password policy: 90 days", "密码已过期检查：正常", + "登录成功 >", "密码错误>", "系统警告 >", "[Warning]#", + } { + t.Run(banner, func(t *testing.T) { + data := []byte("Login:\r\nPassword:\r\n" + banner + "\r\nrouter#\r\nTAIL") + for _, size := range []int{1, 2, 7, 1024, 4096} { + t.Run(fmt.Sprint(size), func(t *testing.T) { + checkBannerStream(t, streamChunks(data, size)) + }) + } + }) + } +} + +func TestLoginStreamEverySplitPreservesTail(t *testing.T) { + data := []byte("Login:\r\nPassword:\r\nPassword expired policy: disabled\r\nrouter#\r\nTAIL") + for split := 0; split <= len(data); split++ { + t.Run(fmt.Sprint(split), func(t *testing.T) { + checkBannerStream(t, [][]byte{data[:split], data[split:]}) + }) + } +} + +func TestLoginStreamFinalMessages(t *testing.T) { + for _, tc := range []struct { + message string + want error + }{ + {"Login incorrect", ErrLoginFailed}, + {"Authentication failed", ErrLoginFailed}, + {"Your password has expired", ErrPasswordChangeRequired}, + {"New password:", ErrPasswordChangeRequired}, + } { + for _, ending := range []string{"", "\r\n"} { + t.Run(tc.message+fmt.Sprintf("/%q", ending), func(t *testing.T) { + conn := &streamTestConn{chunks: streamChunks([]byte("Password:\r\n"+tc.message+ending), 2), lastError: io.EOF} + _, err := NewClientConn(conn, &ClientConfig{Password: "secret", Timeout: time.Second}) + if !errors.Is(err, tc.want) { + t.Fatalf("got %v, want %v", err, tc.want) + } + if conn.writes.String() != "secret\r\n" { + t.Fatalf("writes = %q", conn.writes.String()) + } + }) + } + } +} + +func TestLoginStreamPercentFailureIsNotShell(t *testing.T) { + conn := &streamTestConn{chunks: [][]byte{[]byte("Password:"), []byte("\r\n% Bad passwords\r\n")}} + _, err := NewClientConn(conn, &ClientConfig{Password: "secret", Timeout: time.Second}) + if !errors.Is(err, ErrLoginFailed) { + t.Fatalf("got %v, want ErrLoginFailed", err) + } +} + +func TestLoginStreamLongLineCannotBecomePrompt(t *testing.T) { + data := []byte("Login:\r\nPassword:\r\n" + strings.Repeat("x", 65537) + "router#\r\nrouter#\r\nTAIL") + checkBannerStream(t, streamChunks(data, 4096)) +} + +func TestLoginStreamInteractivePromptsWithoutNewlines(t *testing.T) { + server, raw := net.Pipe() + t.Cleanup(func() { server.Close(); raw.Close() }) + done := make(chan error, 1) + go func() { + defer server.Close() + _ = server.SetDeadline(time.Now().Add(2 * time.Second)) + steps := []struct{ prompt, reply string }{ + {"Lo", ""}, {"gin:", "admin\r\n"}, + {"\x1b[31mPass", ""}, {"word:\x1b[0m", "secret\r\n"}, + {"Press ENTER to continue", "\r\n"}, + } + for _, step := range steps { + if _, err := io.WriteString(server, step.prompt); err != nil { + done <- err + return + } + if step.reply != "" { + got := make([]byte, len(step.reply)) + _, err := io.ReadFull(server, got) + if err != nil { + done <- err + return + } + if string(got) != step.reply { + done <- fmt.Errorf("reply %q, want %q", got, step.reply) + return + } + } + } + _, err := io.WriteString(server, "核心交换机#\r\nALARM") + done <- err + }() + client, err := NewClientConn(raw, &ClientConfig{User: "admin", Password: "secret", Timeout: time.Second}) + if err != nil { + t.Fatal(err) + } + defer client.Close() + output, err := io.ReadAll(client) + if err != nil { + t.Fatal(err) + } + if string(output) != "\r\nALARM" { + t.Fatalf("tail=%q", output) + } + if err := <-done; err != nil { + t.Fatal(err) + } +} + +func TestLoginStreamUnterminatedMessageAtTimeout(t *testing.T) { + server, raw := net.Pipe() + t.Cleanup(func() { server.Close(); raw.Close() }) + done := make(chan struct{}) + go func() { + defer close(done) + defer server.Close() + _ = server.SetDeadline(time.Now().Add(time.Second)) + _, _ = io.WriteString(server, "Your password has expired") + var one [1]byte + _, _ = server.Read(one[:]) + }() + _, err := NewClientConn(raw, &ClientConfig{Password: "secret", Timeout: 50 * time.Millisecond}) + if !errors.Is(err, ErrPasswordChangeRequired) { + t.Fatalf("got %v", err) + } + <-done +} + +func TestLoginStreamColorAndUTF8Fragments(t *testing.T) { + data := []byte("Login:\r\nPassword:\r\n\x1b]0;title\a\x1b[32m核心交换机#\x1b[0m\r\nTAIL") + for _, size := range []int{1, 2, 5, 4096} { + conn := &streamTestConn{chunks: streamChunks(data, size)} + client, err := NewClientConn(conn, &ClientConfig{User: "admin", Password: "secret", Timeout: time.Second}) + if err != nil { + t.Fatal(err) + } + got, err := io.ReadAll(client) + client.Close() + if err != nil { + t.Fatal(err) + } + // A color reset arriving after the prompt is valid session data. + got = bytes.ReplaceAll(got, []byte("\x1b[0m"), nil) + if string(got) != "\r\nTAIL" { + t.Fatalf("chunk %d tail=%q", size, got) + } + } +} + +func FuzzLoginDiagnosticChunking(f *testing.F) { + f.Add(uint16(1), uint8(0), "0") + f.Add(uint16(4096), uint8(1), "disabled") + f.Fuzz(func(t *testing.T, size uint16, choice uint8, suffix string) { + if len(suffix) > 2048 { + suffix = suffix[:2048] + } + suffix = strings.Map(func(r rune) rune { + if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == ' ' { + return r + } + return -1 + }, suffix) + prefixes := []string{"Authentication failed count: ", "Password expired policy: ", "New password policy: "} + data := []byte("Login:\r\nPassword:\r\n" + prefixes[int(choice)%len(prefixes)] + suffix + "\r\nrouter#\r\nTAIL") + checkBannerStream(t, streamChunks(data, int(size)%4096+1)) + }) +} + +func BenchmarkLoginLongBanner(b *testing.B) { + for _, size := range []int{1024, 4096, 65536} { + b.Run(fmt.Sprint(size), func(b *testing.B) { + data := []byte(strings.Repeat("x", size) + "\r\nrouter#") + config, _ := prepareConfig(&ClientConfig{Password: "secret"}) + b.SetBytes(int64(len(data))) + b.ReportAllocs() + for n := 0; n < b.N; n++ { + machine := newLoginMachine(config) + text := newLoginText(config.maxLoginBuffer) + for _, chunk := range streamChunks(data, 4096) { + _, _, _, err := machine.consumeChunk(io.Discard, text, chunk, nil) + if err != nil { + b.Fatal(err) + } + } + } + }) + } +} diff --git a/login_text.go b/login_text.go new file mode 100644 index 0000000..b1c30cf --- /dev/null +++ b/login_text.go @@ -0,0 +1,149 @@ +package tclientlib + +import "unicode/utf8" + +type loginEscapeState uint8 + +const ( + loginEscapeNone loginEscapeState = iota + loginEscapeStart + loginEscapeCSI + loginEscapeOSC + loginEscapeOSCEscape +) + +type loginText struct { + line []byte + maxLine int + pendingCR bool + escape loginEscapeState + overflow bool +} + +func newLoginText(maxLine int) *loginText { + return &loginText{maxLine: maxLine} +} + +// feedByte performs only bounded, linear text normalization. Returned line +// slices alias the scanner and are valid until the next call. CRLF/LF complete +// a line; a bare CR overwrites it. Read/chunk boundaries never complete a line. +func (t *loginText) feedByte(value byte) ([]byte, string) { + if t.consumeEscape(value) { + return nil, "" + } + if value == 0 || value == '\a' { + return nil, "" + } + if value == '\x1b' { + t.escape = loginEscapeStart + return nil, "" + } + if t.pendingCR { + t.pendingCR = false + if value == '\n' { + return t.finishLine(), "\r\n" + } + t.line = t.line[:0] + t.overflow = false + } + switch value { + case '\r': + t.pendingCR = true + case '\n': + return t.finishLine(), "\n" + case '\b', 0x7f: + t.removeLastRune() + case '\t': + t.appendByte(' ') + default: + if value >= 0x20 { + t.appendByte(value) + } + } + return nil, "" +} + +// Incomplete escape sequences survive chunk boundaries. UTF-8 validation is +// performed only on bounded prompt candidates, not on every long-line prefix. +func (t *loginText) current() []byte { + if t.overflow || t.escape != loginEscapeNone { + return nil + } + return t.line +} + +func (t *loginText) consumeEscape(value byte) bool { + switch t.escape { + case loginEscapeNone: + return false + case loginEscapeStart: + switch value { + case '[': + t.escape = loginEscapeCSI + case ']': + t.escape = loginEscapeOSC + default: + t.escape = loginEscapeNone + } + return true + case loginEscapeCSI: + if value >= 0x40 && value <= 0x7e { + t.escape = loginEscapeNone + } + return true + case loginEscapeOSC: + switch value { + case '\a': + t.escape = loginEscapeNone + case '\x1b': + t.escape = loginEscapeOSCEscape + } + return true + case loginEscapeOSCEscape: + if value == '\\' { + t.escape = loginEscapeNone + } else { + t.escape = loginEscapeOSC + } + return true + default: + t.escape = loginEscapeNone + return true + } +} + +func (t *loginText) finishLine() []byte { + line := t.current() + t.line = t.line[:0] + t.overflow = false + return line +} + +func (t *loginText) consumeCurrent() { + t.line = t.line[:0] + t.pendingCR = false + t.overflow = false +} + +func (t *loginText) appendByte(value byte) { + if len(t.line) >= t.maxLine { + // Do not slide a truncated banner's suffix into a synthetic prompt. + // Ignore this logical line until its next reset or line boundary. + t.overflow = true + return + } + if !t.overflow { + t.line = append(t.line, value) + } +} + +func (t *loginText) removeLastRune() { + if len(t.line) == 0 { + return + } + _, size := utf8.DecodeLastRune(t.line) + if size <= 0 { + size = 1 + } + t.line = t.line[:len(t.line)-size] +} From 40d8000eb02ba0464aa7fb6b6a9aa6101bf60661 Mon Sep 17 00:00:00 2001 From: Eric_Lee Date: Tue, 3 Nov 2026 18:36:34 +0800 Subject: [PATCH 2/3] fix: redact all login trace payloads centrally Protect incoming and generated subnegotiation parameters as well as wire/application data. Keep redaction active during stream-wrapper setup and failed-login cleanup. --- trace.go | 37 ++++++++++++++++----- trace_secrecy_test.go | 75 +++++++++++++++++++++++++++++++++++++++++++ transport.go | 28 +++------------- 3 files changed, 109 insertions(+), 31 deletions(-) create mode 100644 trace_secrecy_test.go diff --git a/trace.go b/trace.go index 44998ce..aaceb02 100644 --- a/trace.go +++ b/trace.go @@ -4,13 +4,15 @@ import ( "fmt" "io" "sync" + "sync/atomic" ) const maxTraceBytes = 256 type tracer struct { - writer io.Writer - mu sync.Mutex + writer io.Writer + mu sync.Mutex + redactPayloads atomic.Bool } func newTracer(writer io.Writer) *tracer { @@ -20,6 +22,14 @@ func newTracer(writer io.Writer) *tracer { return &tracer{writer: writer} } +// Redaction lives at the logging boundary so no wire, application, or SB +// payload path can accidentally bypass it during authentication. +func (t *tracer) setRedaction(enabled bool) { + if t != nil { + t.redactPayloads.Store(enabled) + } +} + func (t *tracer) printf(format string, args ...interface{}) { if t == nil { return @@ -33,6 +43,10 @@ func (t *tracer) bytes(scope, direction string, data []byte) { if t == nil || len(data) == 0 { return } + if t.redactPayloads.Load() { + t.redacted(scope, direction, len(data)) + return + } sample, suffix := boundedTraceData(data) t.printf("%s %s len=%d data=% x%s", scope, direction, len(data), sample, suffix) } @@ -53,9 +67,7 @@ func (t *tracer) event(direction string, event telnetEvent) { t.printf("event %s %s %s", direction, telnetCommandName(event.command), telnetOptionName(event.option)) case subnegotiationEvent: - sample, suffix := boundedTraceData(event.parameters) - t.printf("event %s SB %s len=%d data=% x%s", direction, - telnetOptionName(event.option), len(event.parameters), sample, suffix) + t.subnegotiation("event", direction, event.option, event.parameters) case commandEvent: t.printf("event %s %s", direction, telnetCommandName(event.command)) } @@ -66,15 +78,24 @@ func (t *tracer) packet(direction string, packet optionPacket) { return } if packet.command == sb { - sample, suffix := boundedTraceData(packet.parameters) - t.printf("packet %s SB %s len=%d data=% x%s", direction, - telnetOptionName(packet.option), len(packet.parameters), sample, suffix) + t.subnegotiation("packet", direction, packet.option, packet.parameters) return } t.printf("packet %s %s %s", direction, telnetCommandName(packet.command), telnetOptionName(packet.option)) } +func (t *tracer) subnegotiation(kind, direction string, option byte, data []byte) { + if t.redactPayloads.Load() { + t.printf("%s %s SB %s len=%d data=", kind, direction, + telnetOptionName(option), len(data)) + return + } + sample, suffix := boundedTraceData(data) + t.printf("%s %s SB %s len=%d data=% x%s", kind, direction, + telnetOptionName(option), len(data), sample, suffix) +} + func boundedTraceData(data []byte) ([]byte, string) { if len(data) <= maxTraceBytes { return data, "" diff --git a/trace_secrecy_test.go b/trace_secrecy_test.go new file mode 100644 index 0000000..1406d65 --- /dev/null +++ b/trace_secrecy_test.go @@ -0,0 +1,75 @@ +package tclientlib + +import ( + "bytes" + "errors" + "fmt" + "io" + "strings" + "testing" + "time" +) + +func assertNoTracePayload(t *testing.T, output string, values ...string) { + t.Helper() + for _, value := range values { + if strings.Contains(output, value) || strings.Contains(output, fmt.Sprintf("% x", []byte(value))) { + t.Fatalf("trace contains protected payload %q:\n%s", value, output) + } + } +} + +func TestLoginTraceRedactsSubnegotiationParameters(t *testing.T) { + var trace bytes.Buffer + wire := []byte{iac, do, newEnvironment, iac, sb, newEnvironment, 1, iac, se} + wire = append(wire, (optionPacket{command: sb, option: newEnvironment, parameters: []byte("\x00received-secret")}).bytes()...) + wire = append(wire, []byte("Login:\r\nPassword:\r\nrouter#")...) + conn := &streamTestConn{chunks: [][]byte{wire}} + client, err := NewClientConn(conn, &ClientConfig{User: "private-user", Password: "private-pass", Trace: &trace, Timeout: time.Second}) + if err != nil { + t.Fatal(err) + } + defer client.Close() + assertNoTracePayload(t, trace.String(), "private-user", "private-pass", "received-secret") + for _, want := range []string{"event recv SB NEW_ENVIRON", "packet send SB NEW_ENVIRON", "login success", ""} { + if !strings.Contains(trace.String(), want) { + t.Fatalf("missing %q:\n%s", want, trace.String()) + } + } + trace.Reset() + if _, err := client.Write([]byte("show")); err != nil { + t.Fatal(err) + } + if !strings.Contains(trace.String(), "73 68 6f 77") { + t.Fatalf("post-login trace not restored: %s", trace.String()) + } +} + +type traceFlushWriter struct{ io.Writer } + +func (w traceFlushWriter) Close() error { + _, err := w.Write([]byte("cleanup-secret")) + return err +} + +func TestFailedLoginTraceStaysRedactedThroughWrapperClose(t *testing.T) { + var trace bytes.Buffer + conn := &streamTestConn{chunks: [][]byte{[]byte("Password:\r\nLogin incorrect\r\n")}} + _, err := NewClientConn(conn, &ClientConfig{ + Password: "private-pass", Trace: &trace, Timeout: time.Second, + WrapStream: func(r io.Reader, w io.Writer) (io.Reader, io.Writer) { + _, _ = w.Write([]byte("setup-secret")) + return r, traceFlushWriter{w} + }, + }) + if !errors.Is(err, ErrLoginFailed) { + t.Fatalf("got %v", err) + } + if !conn.closed.Load() { + t.Fatal("connection not closed") + } + assertNoTracePayload(t, trace.String(), "private-pass", "setup-secret", "cleanup-secret") + if !strings.Contains(conn.writes.String(), "cleanup-secret") { + t.Fatal("close did not exercise buffered write") + } +} diff --git a/transport.go b/transport.go index 2bed404..d22f0bb 100644 --- a/transport.go +++ b/transport.go @@ -6,7 +6,6 @@ import ( "io" "net" "sync" - "sync/atomic" ) type transport struct { @@ -22,7 +21,6 @@ type transport struct { pending []byte pendingErr error - traceRedacted atomic.Bool localOptions [256]bool remoteOptions [256]bool } @@ -38,7 +36,7 @@ func newTransport(conn net.Conn, config transportConfig) *transport { } func (t *transport) setTraceRedaction(enabled bool) { - t.traceRedacted.Store(enabled) + t.trace.setRedaction(enabled) } func (t *transport) Read(p []byte) (int, error) { @@ -67,17 +65,9 @@ func (t *transport) Read(p []byte) (int, error) { n, readErr := t.conn.Read(t.readBuffer) if n > 0 { raw := t.readBuffer[:n] - if t.traceRedacted.Load() { - t.trace.redacted("wire", "recv", len(raw)) - } else { - t.trace.bytes("wire", "recv", raw) - } + t.trace.bytes("wire", "recv", raw) data, events, decodeErr := t.decoder.feed(raw) - if t.traceRedacted.Load() { - t.trace.redacted("application", "recv", len(data)) - } else { - t.trace.bytes("application", "recv", data) - } + t.trace.bytes("application", "recv", data) t.pending = append(t.pending, data...) for _, event := range events { t.trace.event("recv", event) @@ -101,11 +91,7 @@ func (t *transport) Read(p []byte) (int, error) { } func (t *transport) Write(p []byte) (int, error) { - if t.traceRedacted.Load() { - t.trace.redacted("application", "send", len(p)) - } else { - t.trace.bytes("application", "send", p) - } + t.trace.bytes("application", "send", p) encoded := appendEscapedIAC(make([]byte, 0, len(p)), p) written, err := t.writeRaw(encoded) if err != nil { @@ -307,11 +293,7 @@ func (t *transport) writeRaw(data []byte) (int, error) { for len(data) > 0 { n, err := t.conn.Write(data) if n > 0 { - if t.traceRedacted.Load() { - t.trace.redacted("wire", "send", n) - } else { - t.trace.bytes("wire", "send", data[:n]) - } + t.trace.bytes("wire", "send", data[:n]) } total += n data = data[n:] From 3eb454408ffd607917e718f8e3fdc900ced5faad Mon Sep 17 00:00:00 2001 From: Eric_Lee Date: Tue, 3 Nov 2026 18:37:09 +0800 Subject: [PATCH 3/3] fix: discard oversized subnegotiations without losing framing Continue through the real IAC SE across reads, honor escaped IAC, preserve following application data and valid events, and cover split points with regression and fuzz tests. --- decoder.go | 34 ++++++++++++--- decoder_recovery_test.go | 92 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 119 insertions(+), 7 deletions(-) create mode 100644 decoder_recovery_test.go diff --git a/decoder.go b/decoder.go index 2cc8b31..01aeec1 100644 --- a/decoder.go +++ b/decoder.go @@ -11,6 +11,8 @@ const ( decoderSubnegotiationOption decoderSubnegotiationData decoderSubnegotiationIAC + decoderDiscardSubnegotiation + decoderDiscardSubnegotiationIAC ) type eventType uint8 @@ -43,6 +45,7 @@ func newDecoder(maxSubnegotiation int) *decoder { func (d *decoder) feed(input []byte) ([]byte, []telnetEvent, error) { data := make([]byte, 0, len(input)) events := make([]telnetEvent, 0, 2) + var decodeErr error for _, value := range input { switch d.state { @@ -87,37 +90,54 @@ func (d *decoder) feed(input []byte) ([]byte, []telnetEvent, error) { d.state = decoderSubnegotiationIAC continue } - if err := d.appendSubByte(value); err != nil { - return data, events, err + if err := d.appendSubByte(value); err != nil && decodeErr == nil { + decodeErr = err } case decoderSubnegotiationIAC: switch value { case se: events = append(events, telnetEvent{ - typ: subnegotiationEvent, - option: d.subOption, + typ: subnegotiationEvent, + option: d.subOption, parameters: append([]byte(nil), d.subData...), }) d.subData = d.subData[:0] d.state = decoderData case iac: if err := d.appendSubByte(iac); err != nil { - return data, events, err + if decodeErr == nil { + decodeErr = err + } + continue } d.state = decoderSubnegotiationData default: // Ignore malformed commands inside SB while keeping the stream framed. d.state = decoderSubnegotiationData } + case decoderDiscardSubnegotiation: + if value == iac { + d.state = decoderDiscardSubnegotiationIAC + } + case decoderDiscardSubnegotiationIAC: + // IAC IAC is escaped payload even while discarding. Only the + // actual IAC SE terminator restores the application stream. + if value == se { + d.state = decoderData + } else { + d.state = decoderDiscardSubnegotiation + } } } - return data, events, nil + return data, events, decodeErr } func (d *decoder) appendSubByte(value byte) error { if len(d.subData) >= d.maxSubnegotiation { - d.state = decoderData + // Report once, but continue scanning this chunk and subsequent reads + // through IAC SE. Never expose an oversized SB's tail as terminal data. + d.state = decoderDiscardSubnegotiation d.subData = d.subData[:0] return fmt.Errorf("%w: limit %d", ErrSubnegotiationTooLong, d.maxSubnegotiation) } diff --git a/decoder_recovery_test.go b/decoder_recovery_test.go new file mode 100644 index 0000000..1f41ce6 --- /dev/null +++ b/decoder_recovery_test.go @@ -0,0 +1,92 @@ +package tclientlib + +import ( + "bytes" + "errors" + "fmt" + "testing" +) + +func TestOversizedSubnegotiationKeepsFramingEverySplit(t *testing.T) { + // The escaped IAC followed by SE is payload, not an SB terminator. + wire := []byte{'a', iac, sb, terminalType, 1, 2, 3, 'X', iac, iac, se, 'Y', iac, se, + 'b', iac, will, echoOption, iac, sb, terminalType, 1, iac, se} + for split := 0; split <= len(wire); split++ { + t.Run(fmt.Sprint(split), func(t *testing.T) { + d := newDecoder(2) + var output []byte + var events []telnetEvent + errorsSeen := 0 + for _, part := range [][]byte{wire[:split], wire[split:]} { + data, ev, err := d.feed(part) + output = append(output, data...) + events = append(events, ev...) + if err != nil { + if !errors.Is(err, ErrSubnegotiationTooLong) { + t.Fatal(err) + } + errorsSeen++ + } + if len(d.subData) > 2 { + t.Fatal("unbounded subnegotiation") + } + } + if !bytes.Equal(output, []byte("ab")) { + t.Fatalf("application data = %q", output) + } + if errorsSeen != 1 { + t.Fatalf("errors = %d, want 1", errorsSeen) + } + if len(events) != 2 || events[0].typ != negotiationEvent || events[1].typ != subnegotiationEvent || !bytes.Equal(events[1].parameters, []byte{1}) { + t.Fatalf("events = %#v", events) + } + }) + } +} + +func TestOversizedEscapedIACEntersDiscardState(t *testing.T) { + d := newDecoder(1) + _, _, err := d.feed([]byte{iac, sb, terminalType, 1, iac, iac}) + if !errors.Is(err, ErrSubnegotiationTooLong) { + t.Fatalf("got %v", err) + } + data, ev, err := d.feed([]byte{'X', iac, se, 'z'}) + if err != nil || len(ev) != 0 || string(data) != "z" { + t.Fatalf("data=%q events=%v err=%v", data, ev, err) + } +} + +func FuzzOversizedSubnegotiationRecovery(f *testing.F) { + f.Add([]byte{iac, se, iac, iac}, uint8(1)) + f.Add([]byte("ignored-data"), uint8(7)) + f.Fuzz(func(t *testing.T, payload []byte, size uint8) { + if len(payload) > 4096 { + payload = payload[:4096] + } + wire := []byte{'a', iac, sb, terminalType, 1, 2, 3} + wire = appendEscapedIAC(wire, payload) + wire = append(wire, iac, se, 'b') + d := newDecoder(2) + var output []byte + errorsSeen := 0 + for _, chunk := range streamChunks(wire, int(size)+1) { + data, events, err := d.feed(chunk) + output = append(output, data...) + if len(events) != 0 { + t.Fatalf("oversized SB produced events: %v", events) + } + if err != nil { + if !errors.Is(err, ErrSubnegotiationTooLong) { + t.Fatal(err) + } + errorsSeen++ + } + if len(d.subData) > 2 { + t.Fatal("unbounded SB data") + } + } + if string(output) != "ab" || errorsSeen != 1 { + t.Fatalf("data=%q errors=%d", output, errorsSeen) + } + }) +}