Skip to content

Attribute gated sessions in correlate via the backend port #50

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kilo666mj merged 1 commit into from
Sep 27, 2026
Merged

Attribute gated sessions in correlate via the backend port #50

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
185 changes: 165 additions & 20 deletions correlate.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 <path>] [--window <duration>] <fingerprint>")
fatalf("usage: correlate [--log <path>] [--gate-log <path>] [--window <duration>] <fingerprint>")
}

st, err := NewStore(*dbPath)
Expand All @@ -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)
}
Expand All @@ -59,27 +90,34 @@ 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),
)
}
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
Expand All @@ -102,27 +140,130 @@ 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
}
}
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 <ip> port <n>"
// 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") &&
Expand Down Expand Up @@ -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 <name>" (OpenSSH) or "for invalid <name>".
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], " ,")
}
Expand Down
126 changes: 126 additions & 0 deletions correlate_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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")
}
}
Loading
Loading