Lab 2: Build a production-style log triage pipeline

This lab composes the tools from Part II into a real production diagnostic. The output is a script that triage an SSH log and identify the top source IPs that failed to authenticate.

Objective

Build a shell pipeline that reads a Linux system’s SSH authentication log, extracts the source IP of every failed login attempt, counts them, and produces the top 10. The pipeline must:

Survive a log file with 100 MB of entries.
Not break on rotated (*.1, *.gz) log files.
Save intermediate output for offline inspection.
Fail loudly (exit non-zero) if any stage fails, except for the “no matches” case.

Architecture

flowchart LR
  L["SSH log files<br/>/var/log/auth.log, /var/log/secure,<br/>journalctl -u ssh"]
  R["Rotated logs<br/>*.1, *.gz"]
  R --> Z["zcat / zless"]
  L --> C["cat"]
  Z --> C
  C --> G["grep -E 'Failed password|Invalid user'"]
  G --> A["awk + sort + uniq -c"]
  A --> T["tee /tmp/ssh-failures.log"]
  T --> S["sort -rn | head"]
  T --> O["Persisted intermediate"]

Requirements

A Linux host with at least a few dozen SSH login attempts recorded in the journal (journalctl -u ssh on systemd systems).
For a real-world test: a host that has been on the internet for at least a week, with a /var/log/auth.log (Debian-family) or /var/log/secure (RHEL-family).
bash, grep, awk, sort, uniq, head, tee, zcat.

Scenario

You have been paged at 02:00 because SSH login latency is up. You suspect a brute-force attempt from a botnet. Your task: in ten minutes, identify the top source IPs of failed login attempts, save the evidence, and report.

Tasks

Task 1: Read the journal and find the failure pattern

Start with the live journal:

journalctl -u ssh --since "1 day ago" --no-pager | head -20

Look for the line pattern. On Debian-family:

Aug  9 03:12:01 host sshd[12345]: Failed password for invalid user admin from 203.0.113.45 port 51234 ssh2

On RHEL-family the message is slightly different (Failed password versus Failed none); the pipeline below handles both.

Task 2: Extract the source IP with awk

journalctl -u ssh --since "1 day ago" --no-pager \
  | awk '/Failed password|Invalid user/ { for (i=1; i<=NF; i++) if ($i == "from") print $(i+1) }' \
  | sort | uniq -c | sort -rn | head -10

Walk through what the awk does:

The outer /Failed password|Invalid user/ matches only relevant log lines.
The inner for scans every field of the matched line.
if ($i == "from") finds the literal word “from”.
print $(i+1) prints the next field — the IP address.

Task 3: Handle compressed rotated logs

Most production logs are rotated and gzipped. The pipeline needs to read both /var/log/auth.log and /var/log/auth.log.*.gz. A small loop:

{
  for f in /var/log/auth.log /var/log/auth.log.*.gz; do
    [ -e "$f" ] || continue
    case "$f" in
      *.gz) zcat "$f" ;;
      *)    cat "$f" ;;
    esac
  done
} | awk '/Failed password|Invalid user/ { for (i=1; i<=NF; i++) if ($i == "from") print $(i+1) }' \
  | sort | uniq -c | sort -rn | head -10

This is the canonical “read all rotated logs in time order” pattern.

Task 4: Save intermediate state with tee

The pipeline above is informative but ephemeral. If the run fails or you get interrupted, you lose the evidence. Add tee:

{
  for f in /var/log/auth.log /var/log/auth.log.*.gz; do
    [ -e "$f" ] || continue
    case "$f" in
      *.gz) zcat "$f" ;;
      *)    cat "$f" ;;
    esac
  done
} | awk '/Failed password|Invalid user/ { for (i=1; i<=NF; i++) if ($i == "from") print $(i+1) }' \
  | sort | uniq -c | sort -rn \
  | tee /tmp/ssh-failures-$(date +%Y%m%d-%H%M%S).log \
  | head -10

tee writes the full ranked list to a timestamped file in /tmp and also passes it on to head. The saved file is your evidence.

Task 5: Fail loudly

Wrap the pipeline as a script and add set -euo pipefail:

#!/usr/bin/env bash
set -euo pipefail

OUT="/tmp/ssh-failures-$(date +%Y%m%d-%H%M%S).log"
trap 'rm -f "$OUT"' EXIT

{
  for f in /var/log/auth.log /var/log/auth.log.*.gz; do
    [ -e "$f" ] || continue
    case "$f" in
      *.gz) zcat "$f" ;;
      *)    cat "$f" ;;
    esac
  done
} | awk '/Failed password|Invalid user/ { for (i=1; i<=NF; i++) if ($i == "from") print $(i+1) }' \
  | sort | uniq -c | sort -rn \
  | tee "$OUT" \
  | head -10

set -e — exit on any error.
set -u — error on undefined variables.
set -o pipefail — pipeline returns the rightmost non-zero exit.

The trap removes the intermediate file on clean exit. For incident evidence you may want to keep it — drop the trap or move the file somewhere persistent.

Task 6: Test against synthetic input

Don’t wait for a real incident to validate the script. Generate synthetic data:

mkdir -p /tmp/ssh-lab
LOG=/tmp/ssh-lab/auth.log
> "$LOG"

for i in $(seq 1 1000); do
  ip="10.0.$((RANDOM % 5)).$((RANDOM % 250))"
  echo "Aug  9 03:12:01 host sshd[$i]: Failed password for invalid user admin from $ip port 51234 ssh2" >> "$LOG"
done

# Run the script with the synthetic log
LOG_DIR=/tmp/ssh-lab bash ./log-triage.sh

Confirm the output contains the top 10 source IPs by frequency.

Validation

The lab is complete when:

The script reads from /var/log/auth.log and /var/log/auth.log.*.gz.
The script saves an intermediate file in /tmp with the full ranked list.
The script prints the top 10 to stdout.
The script exits 0 on a normal log, exits non-zero if awk or sort fails.
The script handles “no matches” without exiting non-zero (the pipeline should not fail just because no failed logins were found).

Expected outcome

A reusable script in /tmp/log-triage.sh that you can run on any host with SSH logs. The intermediate file in /tmp shows the full evidence trail, useful for incident review.

Troubleshooting

“Permission denied” reading /var/log/secure — RHEL-family protects the secure log. Run with sudo or delegate via logadm/ACL.
“command not found: zcat” — install gzip (Debian: apt install gzip).
“no matches” but the host has many failed logins — your regex doesn’t match your distribution’s log format. Look at one log line with head -1 and adjust.
Output is empty because set -o pipefail triggered on the “no matches” branch — wrap the awk in { … } || true if “no matches” is acceptable; better, structure the script so the pipeline ends in head which always succeeds.

Cleanup

rm -rf /tmp/ssh-lab /tmp/log-triage.sh /tmp/ssh-failures-*.log

What you learned

You composed grep, awk, sort, uniq, head, tee, and shell quoting into a production-style diagnostic. The same shape — … | tee <evidence> | <action> — works for log triage, performance investigation, and security audits. The discipline is to save the evidence before the conclusion.

Objectives

Prerequisites