diff --git a/correlate.go b/correlate.go index 42ac872..c9d8297 100644 --- a/correlate.go +++ b/correlate.go @@ -5,33 +5,57 @@ import ( "errors" "flag" "fmt" + "net" "os" + "regexp" "sort" + "strconv" "strings" "text/tabwriter" "time" ) -const defaultAuthLog = "/var/log/auth.log" +const ( + defaultAuthLog = "/var/log/auth.log" + // sshgate logs to the journal; rsyslog copies it to syslog on Debian. + defaultGateLog = "/var/log/syslog" +) type logMatch struct { + when time.Time + ip string + source string + user string + line string +} + +// gateConnect is one sshgate CONNECTED line: the client address and the local +// port of the backend socket, which is the port sshd logs for the session. +type gateConnect struct { when time.Time + host string ip string - user string - line string + port int } +var ( + gateConnectRE = regexp.MustCompile(`\[([^\]]+)\] CONNECTED (\S+) backend=\S+ local=(\S+)`) + sshdSourceRE = regexp.MustCompile(`from (\S+) port (\d+)`) +) + func cmdCorrelate(args []string) { fs := flag.NewFlagSet("correlate", flag.ExitOnError) dbPath := fs.String("db", defaultDB, "database path") logPath := fs.String("log", defaultAuthLog, "sshd log path") - window := fs.Duration("window", 2*time.Minute, "time window around first/last seen") + gateLog := fs.String("gate-log", defaultGateLog, "log with sshgate CONNECTED lines, used to attribute sessions sshd logs as loopback") + window := fs.Duration("window", 2*time.Minute, "time window around first/last seen for direct connections") + joinWindow := fs.Duration("join-window", 10*time.Second, "maximum time between an sshgate CONNECTED line and the sshd line joined to it") limit := fs.Int("limit", 100, "maximum matches to print") if err := fs.Parse(args); err != nil { fatalf("parse correlate options: %v", err) } if fs.NArg() != 1 { - fatalf("usage: correlate [--log ] [--window ] ") + fatalf("usage: correlate [--log ] [--gate-log ] [--window ] ") } st, err := NewStore(*dbPath) @@ -50,7 +74,14 @@ func cmdCorrelate(args []string) { fatalf("fingerprint %s has no IPs to correlate", fp) } - matches, err := correlateSSHDLog(*logPath, entry, *window, *limit) + gates, err := readGateConnects(*gateLog, fp, entry.LastSeen.Time) + if err != nil { + if !errors.Is(err, os.ErrNotExist) { + fatalf("read sshgate log: %v", err) + } + fmt.Fprintf(os.Stderr, "warning: %v; sessions sshd logs as loopback cannot be attributed\n", err) + } + matches, err := correlateSSHDLog(*logPath, entry, *window, *limit, gates, *joinWindow) if err != nil { fatalf("correlate sshd log: %v", err) } @@ -59,19 +90,21 @@ func cmdCorrelate(args []string) { if entry.Label != "" { fmt.Printf("label: %s\n", entry.Label) } - fmt.Printf("window: +/- %s around first_seen=%s and last_seen=%s\n", + fmt.Printf("window: +/- %s around first_seen=%s and last_seen=%s (direct connections)\n", window.String(), formatTime(entry.FirstSeen), formatTime(entry.LastSeen)) + fmt.Printf("gated sessions: %d CONNECTED line(s) in %s\n", len(gates), *gateLog) if len(matches) == 0 { fmt.Println("no matching sshd log lines found") return } w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) - writeOrFatal(w, "TIME\tIP\tUSER\tLINE\n") + writeOrFatal(w, "TIME\tIP\tSOURCE\tUSER\tLINE\n") for _, m := range matches { - writeOrFatal(w, "%s\t%s\t%s\t%s\n", + writeOrFatal(w, "%s\t%s\t%s\t%s\t%s\n", m.when.Format("2006-01-02 15:04:05"), m.ip, + m.source, valueOrDash(sanitizeDisplay(m.user)), sanitizeDisplay(m.line), ) @@ -79,7 +112,12 @@ func cmdCorrelate(args []string) { flushOrFatal(w) } -func correlateSSHDLog(path string, entry Entry, window time.Duration, limit int) (_ []logMatch, err error) { +// correlateSSHDLog reports sshd lines for the fingerprint's sessions. When +// sshd listens behind sshgate it logs every client as loopback, so those +// lines are joined to the gate's CONNECTED line for the same host and port; +// lines from a real client address are matched against the fingerprint's +// known IPs near its first/last seen times. +func correlateSSHDLog(path string, entry Entry, window time.Duration, limit int, gates []gateConnect, joinWindow time.Duration) (_ []logMatch, err error) { f, err := os.Open(path) if err != nil { return nil, err @@ -102,20 +140,25 @@ func correlateSSHDLog(path string, entry Entry, window time.Duration, limit int) if !isUsefulSSHDLog(line) { continue } - ip, ok := lineContainsAnyIP(line, ips) + when, ok := parseLogTime(line, entry.LastSeen.Time) if !ok { continue } - when, ok := parseLogTime(line, entry.LastSeen.Time) - if !ok || !withinCorrelationWindows(when, entry, window) { - continue + m := logMatch{when: when, user: extractSSHUser(line), line: line} + if src, port, ok := sshdSource(line); ok && isLoopbackAddr(src) { + g, ok := matchGateConnect(gates, logHost(line), port, when, joinWindow) + if !ok { + continue + } + m.ip, m.source = g.ip, fmt.Sprintf("gate:%d", port) + } else { + ip, ok := lineContainsAnyIP(line, ips) + if !ok || !withinCorrelationWindows(when, entry, window) { + continue + } + m.ip, m.source = ip, "direct" } - matches = append(matches, logMatch{ - when: when, - ip: ip, - user: extractSSHUser(line), - line: line, - }) + matches = append(matches, m) if limit > 0 && len(matches) >= limit { break } @@ -123,6 +166,104 @@ func correlateSSHDLog(path string, entry Entry, window time.Duration, limit int) return matches, scanner.Err() } +// readGateConnects returns the CONNECTED lines logged for fingerprint. +func readGateConnects(path, fingerprint string, ref time.Time) (_ []gateConnect, err error) { + f, err := os.Open(path) + if err != nil { + return nil, err + } + defer func() { + if closeErr := f.Close(); closeErr != nil { + err = errors.Join(err, fmt.Errorf("close sshgate log: %w", closeErr)) + } + }() + var gates []gateConnect + scanner := bufio.NewScanner(f) + scanner.Buffer(make([]byte, 64*1024), 1024*1024) + for scanner.Scan() { + line := scanner.Text() + if !strings.Contains(line, " CONNECTED "+fingerprint+" ") { + continue + } + m := gateConnectRE.FindStringSubmatch(line) + if m == nil || m[2] != fingerprint { + continue + } + _, portText, err := net.SplitHostPort(m[3]) + if err != nil { + continue + } + port, err := strconv.Atoi(portText) + if err != nil { + continue + } + when, ok := parseLogTime(line, ref) + if !ok { + continue + } + gates = append(gates, gateConnect{when: when, host: logHost(line), ip: m[1], port: port}) + } + return gates, scanner.Err() +} + +// matchGateConnect returns the CONNECTED line closest in time to an sshd line +// with the same host and port. Local ports are reused, so only lines within +// joinWindow count. +func matchGateConnect(gates []gateConnect, host string, port int, when time.Time, joinWindow time.Duration) (gateConnect, bool) { + var ( + best gateConnect + found bool + gap time.Duration + ) + for _, g := range gates { + if g.port != port || (host != "" && g.host != "" && g.host != host) { + continue + } + d := when.Sub(g.when) + if d < 0 { + d = -d + } + if d > joinWindow || (found && d >= gap) { + continue + } + best, found, gap = g, true, d + } + return best, found +} + +// sshdSource returns the address and port from an sshd "from port " +// clause. +func sshdSource(line string) (string, int, bool) { + m := sshdSourceRE.FindStringSubmatch(line) + if m == nil { + return "", 0, false + } + port, err := strconv.Atoi(m[2]) + if err != nil { + return "", 0, false + } + return m[1], port, true +} + +func isLoopbackAddr(s string) bool { + ip := net.ParseIP(s) + return ip != nil && ip.IsLoopback() +} + +// logHost returns the host field of an RFC 3339 or classic syslog line. +func logHost(line string) string { + fields := strings.Fields(line) + if len(fields) > 1 { + if _, err := time.Parse(time.RFC3339Nano, fields[0]); err == nil { + return fields[1] + } + } + if len(fields) > 3 { + return fields[3] + } + return "" +} + func isUsefulSSHDLog(line string) bool { lower := strings.ToLower(line) return strings.Contains(lower, "sshd") && @@ -185,6 +326,10 @@ func extractSSHUser(line string) string { if i+1 < len(fields) && fields[i+1] != "invalid" { return strings.Trim(fields[i+1], " ,") } + // "for invalid user " (OpenSSH) or "for invalid ". + if i+3 < len(fields) && fields[i+1] == "invalid" && fields[i+2] == "user" { + return strings.Trim(fields[i+3], " ,") + } if i+2 < len(fields) && fields[i+1] == "invalid" { return strings.Trim(fields[i+2], " ,") } diff --git a/correlate_test.go b/correlate_test.go new file mode 100644 index 0000000..4bba777 --- /dev/null +++ b/correlate_test.go @@ -0,0 +1,126 @@ +package main + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/kilo666mj/gatekit/store" +) + +const testFP = "aaf62b02afeaa8df0687aa49b07825f8" + +func writeLog(t *testing.T, name string, lines ...string) string { + t.Helper() + path := filepath.Join(t.TempDir(), name) + if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil { + t.Fatal(err) + } + return path +} + +func testEntry(t *testing.T, first, last string, ips ...string) Entry { + t.Helper() + parse := func(s string) store.Time { + v, err := time.Parse(time.RFC3339, s) + if err != nil { + t.Fatal(err) + } + return store.Time{Time: v} + } + return Entry{Fingerprint: testFP, FirstSeen: parse(first), LastSeen: parse(last), IPs: ips} +} + +// Lines as sshd and sshgate write them on a host where sshd listens only on +// 127.0.0.1:22 behind sshgate (RFC 3339 rsyslog format, as on Debian 13). +func TestCorrelateJoinsGatedSessionsOnBackendPort(t *testing.T) { + gateLog := writeLog(t, "syslog", + "2026-09-27T10:20:14.100000+02:00 mx sshgate[3348374]: 2026/09/27 10:20:14 [203.0.113.160] CONNECTED "+testFP+" backend=127.0.0.1:22 local=127.0.0.1:40732", + // Another client's connection must not be attributed to this fingerprint. + "2026-09-27T10:20:30.100000+02:00 mx sshgate[3348374]: 2026/09/27 10:20:30 [198.51.100.9] CONNECTED 1111111111111111111111111111111f backend=127.0.0.1:22 local=127.0.0.1:40800", + ) + authLog := writeLog(t, "auth.log", + "2026-09-27T10:20:14.300000+02:00 mx sshd-session[3349342]: Accepted publickey for michael from 127.0.0.1 port 40732 ssh2: RSA SHA256:dy7FV", + "2026-09-27T10:20:30.300000+02:00 mx sshd-session[3349400]: Accepted publickey for other from 127.0.0.1 port 40800 ssh2: RSA SHA256:zzz", + // Port 40732 reused much later by a connection with no CONNECTED line. + "2026-09-27T11:45:00.000000+02:00 mx sshd-session[3349999]: Accepted publickey for root from 127.0.0.1 port 40732 ssh2: RSA SHA256:yyy", + ) + entry := testEntry(t, "2026-09-27T08:00:00Z", "2026-09-27T08:20:14Z", "203.0.113.160") + + gates, err := readGateConnects(gateLog, testFP, entry.LastSeen.Time) + if err != nil { + t.Fatal(err) + } + if len(gates) != 1 || gates[0].port != 40732 || gates[0].ip != "203.0.113.160" || gates[0].host != "mx" { + t.Fatalf("gates = %+v", gates) + } + matches, err := correlateSSHDLog(authLog, entry, 2*time.Minute, 100, gates, 10*time.Second) + if err != nil { + t.Fatal(err) + } + if len(matches) != 1 { + t.Fatalf("matches = %+v", matches) + } + m := matches[0] + if m.ip != "203.0.113.160" || m.source != "gate:40732" || m.user != "michael" { + t.Fatalf("match = %+v", m) + } +} + +func TestCorrelateGateJoinRequiresSameHost(t *testing.T) { + gates := []gateConnect{{when: time.Date(2026, 9, 27, 8, 20, 14, 0, time.UTC), host: "snuffles", ip: "203.0.113.160", port: 40732}} + authLog := writeLog(t, "auth.log", + "2026-09-27T10:20:14.300000+02:00 mx sshd[1]: Accepted publickey for michael from 127.0.0.1 port 40732 ssh2", + ) + entry := testEntry(t, "2026-09-27T08:00:00Z", "2026-09-27T08:20:14Z", "203.0.113.160") + matches, err := correlateSSHDLog(authLog, entry, 2*time.Minute, 100, gates, 10*time.Second) + if err != nil { + t.Fatal(err) + } + if len(matches) != 0 { + t.Fatalf("joined across hosts: %+v", matches) + } +} + +func TestCorrelateDirectConnectionsStillMatchByIP(t *testing.T) { + authLog := writeLog(t, "auth.log", + // Classic syslog format, and the older sshd process name. + "Sep 27 10:20:14 mx sshd[1234]: Failed password for invalid user admin from 203.0.113.160 port 5555 ssh2", + "Sep 27 10:21:00 mx sshd[1234]: Accepted publickey for michael from 198.51.100.1 port 5556 ssh2", + // Outside the window around first/last seen. + "Sep 27 16:00:00 mx sshd[1234]: Accepted publickey for michael from 203.0.113.160 port 5557 ssh2", + ) + local := time.Date(2026, 9, 27, 10, 20, 0, 0, time.Local) + entry := Entry{Fingerprint: testFP, FirstSeen: store.Time{Time: local}, LastSeen: store.Time{Time: local}, IPs: []string{"203.0.113.160"}} + matches, err := correlateSSHDLog(authLog, entry, 2*time.Minute, 100, nil, 10*time.Second) + if err != nil { + t.Fatal(err) + } + if len(matches) != 1 || matches[0].source != "direct" || matches[0].ip != "203.0.113.160" || matches[0].user != "admin" { + t.Fatalf("matches = %+v", matches) + } +} + +func TestMatchGateConnectPicksNearestWithinWindow(t *testing.T) { + at := time.Date(2026, 9, 27, 8, 0, 0, 0, time.UTC) + gates := []gateConnect{ + {when: at.Add(-8 * time.Second), host: "mx", ip: "198.51.100.1", port: 40000}, + {when: at.Add(-1 * time.Second), host: "mx", ip: "198.51.100.2", port: 40000}, + {when: at.Add(-30 * time.Second), host: "mx", ip: "198.51.100.3", port: 40000}, + } + g, ok := matchGateConnect(gates, "mx", 40000, at, 10*time.Second) + if !ok || g.ip != "198.51.100.2" { + t.Fatalf("got %+v %v", g, ok) + } + if _, ok := matchGateConnect(gates[2:], "mx", 40000, at, 10*time.Second); ok { + t.Fatal("matched a connection outside the join window") + } +} + +func TestIsUsefulSSHDLogAcceptsSSHDSession(t *testing.T) { + if !isUsefulSSHDLog("2026-09-27T10:20:14+02:00 mx sshd-session[1]: Accepted publickey for michael from 127.0.0.1 port 1 ssh2") { + t.Fatal("sshd-session line rejected") + } +} diff --git a/docs/operations.md b/docs/operations.md index fa7d39f..d129de3 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -138,28 +138,34 @@ host does not match the server certificate. ## Correlate -`correlate` matches a fingerprint's known source IPs against `sshd` log lines -near the fingerprint's first/last seen timestamps: +`correlate` lists the `sshd` log lines for a fingerprint's sessions: ```bash -sshgate correlate --db ./sshgate.db --log /var/log/auth.log +sshgate correlate --db ./sshgate.db --log /var/log/auth.log --gate-log /var/log/syslog ``` -Use `--log /var/log/secure` on distributions that write SSH authentication -events there. Use `--window 5m` to widen the matching window. - When `sshd` only listens behind `sshgate`, it logs every client as the gate -(for example `Accepted publickey for alice from 127.0.0.1 port 54321`), so -source IPs do not match. Each connection forwarded to the backend logs a -`CONNECTED` line with the local address of the backend socket: +(for example `Accepted publickey for alice from 127.0.0.1 port 54321`). Each +connection forwarded to the backend logs a `CONNECTED` line with the local +address of the backend socket: ```text [203.0.113.7] CONNECTED backend=127.0.0.1:22 local=127.0.0.1:54321 ``` -The `local` port is the port `sshd` reports, so joining the two lines on host, -port, and a short time window recovers the client IP and fingerprint for each -authenticated session. +The `local` port is the port `sshd` reports, so `correlate` joins each loopback +`sshd` line to the fingerprint's `CONNECTED` line with the same host and port +that is nearest in time within `--join-window` (default `10s`), and reports the +client address from the gate. These rows show `SOURCE gate:`. `sshgate` +logs to the journal; `--gate-log` defaults to `/var/log/syslog`, where rsyslog +copies it on Debian. If that file is missing, `correlate` warns and reports +direct connections only. + +`sshd` lines from a real client address (connections that reached `sshd` +directly) are matched against the fingerprint's known source IPs near its +first/last seen timestamps and show `SOURCE direct`. Use `--window 5m` to widen +that window, and `--log /var/log/secure` on distributions that write SSH +authentication events there. ## Troubleshooting