grep -c "Accepted publickey" /var/log/auth.log
ls -la /home/*/

# Install
apt-get install -y auditd audispd-plugins
systemctl enable --now auditd

# /etc/audit/rules.d/vps_ssh_audit.rules
# Capture SSH login/logout (PAM auth success events)
-w /var/log/auth.log -p wa -k ssh_auth

# Capture all shell command executions (execve)
-a always,exit -F arch=b64 -S execve -F exit=0 -k vps_cmd

# Capture file writes (watch key directories only)
-w /root/.ssh -w /etc -w /opt -w /srv -w /var/www -p wa -k vps_file

# Capture outbound connections
-a always,exit -F arch=b64 -S connect -F exit=0 -k vps_net

# Load rules
auditctl -R /etc/audit/rules.d/vps_ssh_audit.rules

# /opt/ssh_audit/pam_login.sh
#!/bin/bash
echo "$(date -Iseconds) LOGIN user=$PAM_USER ip=$PAM_RHOST uid=$PAM_UID" >> /var/log/ssh_session.log

#!/usr/bin/env python3
"""
collector.py — SSH session collection daemon
Tails auditd events, parses them, and writes to SQLite
"""

import os
import re
import sqlite3
import subprocess
import json
import time
import logging
from datetime import datetime
from pathlib import Path

DB_DIR = Path("/opt/ssh_audit/data")
DB_DIR.mkdir(parents=True, exist_ok=True)

def get_db_path() -> str:
    today = datetime.now().strftime("%Y%m%d")
    return str(DB_DIR / f"ssh_audit_{today}.db")

def init_schema(db_path: str):
    conn = sqlite3.connect(db_path)
    conn.executescript("""
    CREATE TABLE IF NOT EXISTS sessions (
        session_id    TEXT PRIMARY KEY,
        user          TEXT,
        src_ip        TEXT,
        login_time    TEXT,
        logout_time   TEXT,
        duration_sec  INTEGER DEFAULT 0,
        cmd_count     INTEGER DEFAULT 0
    );
    CREATE TABLE IF NOT EXISTS commands (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        session_id    TEXT,
        ts            TEXT,
        user          TEXT,
        cmd           TEXT,
        exit_code     INTEGER
    );
    CREATE TABLE IF NOT EXISTS file_writes (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        session_id    TEXT,
        ts            TEXT,
        user          TEXT,
        filepath      TEXT,
        operation     TEXT
    );
    CREATE TABLE IF NOT EXISTS net_outbound (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        session_id    TEXT,
        ts            TEXT,
        user          TEXT,
        dst_ip        TEXT,
        dst_port      INTEGER,
        proto         TEXT
    );
    CREATE INDEX IF NOT EXISTS idx_cmd_session ON commands(session_id);
    """)
    conn.commit()
    conn.close()

# ── Parse auditd raw lines ──────────────────────────────────────────────

def parse_audit_line(line: str):
    m = re.match(
        r"^\s*type=(\w+)\s+msg=audit\((\d+)\.\d+\):.*?a\d+=\(?(\d+)\"?\s*(.*)$",
        line,
        re.DOTALL,
    )
    return (m.group(1), m.group(0)) if m else None

def extract_execve(payload: str) -> dict | None:
    m = re.search(r'a0=(\d+).*?a1=(\d+).*?cmdline="([^"]*)"', payload)
    return {"cmd": m.group(3)} if m else None

def extract_connect(payload: str) -> dict | None:
    m = re.search(r'addr=(\d+):\d+', payload)
    if not m:
        return None
    dst = m.group(1)
    ip = ".".join(
        str(int(dst[i : i + 2], 16)) for i in range(6, -2, -2)
    )
    return {"dst_ip": ip, "dst_port": 0}

# ── Main collection loop ───────────────────────────────────────────────

SESSIONS = {}

def main():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
    )
    db = get_db_path()
    init_schema(db)
    conn = sqlite3.connect(db)
    logging.info(f"Collector started, DB={db}")

    tail_proc = subprocess.Popen(
        ["tail", "-F", "/var/log/audit/audit.log"],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        text=True,
    )

    for line in tail_proc.stdout:
        line = line.strip()
        if not line:
            continue

        if "pam_login.sh" in line:
            m = re.search(r"user=(\S+)\s+ip=(\S+)", line)
            if m:
                user, ip = m.group(1), m.group(2)
                sid = f"{user}_{int(time.time())}"
                SESSIONS[user] = {"session_id": sid, "src_ip": ip}
                conn.execute(
                    "INSERT OR REPLACE INTO sessions VALUES (?,?,?,?,?,0,0)",
                    (sid, user, ip, datetime.now().isoformat(), None),
                )
                conn.commit()

        elif "execve" in line:
            ev = extract_execve(line)
            if ev:
                conn.execute(
                    "INSERT INTO commands VALUES (?,?,?,?,?,?)",
                    (None, SESSIONS.get("_", {}).get("session_id", "unknown"),
                     datetime.now().isoformat(), "_", ev["cmd"], 0),
                )
                conn.commit()

    tail_proc.terminate()
    conn.close()

if __name__ == "__main__":
    main()

# baseline.py — sliding window aggregation, every 15 minutes

import sqlite3
from datetime import datetime
from pathlib import Path
import json

DB_DIR = Path("/opt/ssh_audit/data")

def compute_user_baseline(user: str, window_days: int = 14) -> dict:
    db = get_db_path_for_range(window_days)
    conn = sqlite3.connect(db)

    # ① Command entropy: distinct / total (high = varied ops, low = repetitive)
    row = conn.execute(
        "SELECT COUNT(*), COUNT(DISTINCT cmd) FROM commands WHERE user=?",
        (user,),
    ).fetchone()
    total, distinct = row
    cmd_entropy = distinct / total if total > 0 else 0.0

    # ② File write rate: avg daily writes
    row = conn.execute(
        "SELECT COUNT(*) FROM file_writes WHERE user=?",
        (user,),
    ).fetchone()
    file_write_rate = row[0] / window_days

    # ③ Outbound target set (deduped IPs)
    rows = conn.execute(
        "SELECT DISTINCT dst_ip FROM net_outbound WHERE user=?",
        (user,),
    ).fetchall()
    net_targets = [r[0] for r in rows]

    # ④ Active hour distribution (24-bin histogram)
    rows = conn.execute(
        "SELECT login_time FROM sessions WHERE user=?",
        (user,),
    ).fetchall()
    hour_bins = [0] * 24
    for r in rows:
        try:
            hour_bins[int(r[0][11:13])] += 1
        except (IndexError, ValueError):
            pass

    conn.close()
    return {
        "user": user,
        "cmd_entropy": round(cmd_entropy, 4),
        "file_write_rate": round(file_write_rate, 2),
        "net_target_count": len(net_targets),
        "net_targets": net_targets[:20],
        "hour_distribution": hour_bins,
        "window_days": window_days,
    }

BASELINE_DB = Path("/opt/ssh_audit/baseline.db")

def save_baseline(baseline: dict):
    conn = sqlite3.connect(BASELINE_DB)
    conn.execute(
        """CREATE TABLE IF NOT EXISTS user_baselines (
            user TEXT, computed_at TEXT,
            cmd_entropy REAL, file_write_rate REAL,
            net_target_count INTEGER, net_targets TEXT,
            hour_distribution TEXT, window_days INTEGER,
            PRIMARY KEY (user, computed_at)
        )"""
    )
    conn.execute(
        "INSERT INTO user_baselines VALUES (?,?,?,?,?,?,?,?)",
        (
            baseline["user"],
            datetime.now().isoformat(),
            baseline["cmd_entropy"],
            baseline["file_write_rate"],
            baseline["net_target_count"],
            json.dumps(baseline["net_targets"]),
            json.dumps(baseline["hour_distribution"]),
            baseline["window_days"],
        ),
    )
    conn.commit()
    conn.close()

# risk_rules.py — static rule engine, no LLM dependency

import re
from dataclasses import dataclass


@dataclass
class RiskEvent:
    user: str
    cmd: str
    ts: str
    risk_level: str   # 'HIGH' / 'MEDIUM' / 'LOW'
    reason: str


def evaluate_risk(user: str, cmd: str, ts: str) -> list[RiskEvent]:
    events: list[RiskEvent] = []
    c = cmd.lower()

    # ── HIGH: credential / key operations ──────────────────────────────
    HIGH_PATTERNS = [
        (r"(?<=\s)cat\s+/etc/shadow", "Reading /etc/shadow (password hashes)"),
        (r"cp\s+\S*\.ssh/", "Copying SSH key directory"),
        (r"rsync.*\.env\b", "Syncing .env file (contains API keys)"),
        (r"git\s+push.*--force", "Force push (may overwrite others' commits)"),
        (r"(rm|sudo\s+rm)\s+-rf\s+/", "Recursive delete from root"),
        (r"chmod\s+777\s+/\S+/", "Dangerous world-writable permissions"),
    ]
    for pattern, reason in HIGH_PATTERNS:
        if re.search(pattern, c):
            events.append(RiskEvent(user, cmd, ts, "HIGH", reason))

    # ── MEDIUM: data exfiltration patterns ────────────────────────────
    MEDIUM_PATTERNS = [
        (r"(scp|rsync|curl|wget|ftp)\s+\S+\s+@\S+", "Transferring files to external destination"),
        (r"tar\s+-czf\s+/tmp/\S+\s+/var", "Packing sensitive dir to /tmp (exfiltratable)"),
        (r"nc\s+\S+\s+\d+", "netcat outbound connection (unusual channel)"),
    ]
    for pattern, reason in MEDIUM_PATTERNS:
        if re.search(pattern, c):
            events.append(RiskEvent(user, cmd, ts, "MEDIUM", reason))

    # ── LOW: suspicious but common ─────────────────────────────────────
    if re.search(r"history\s*-c|shred\s+", c):
        events.append(RiskEvent(user, cmd, ts, "LOW", "Clearing shell history / secure wipe"))

    return events

def on_risk_event(ev: RiskEvent, baseline: dict):
    msg = (
        f"🚨 SSH Audit Alert [{ev.risk_level}]\n"
        f"User: {ev.user}\n"
        f"Time: {ev.ts}\n"
        f"Command: {ev.cmd[:120]}\n"
        f"Reason: {ev.reason}\n"
        f"Baseline deviation: {format_baseline_deviation(ev.user, baseline)}"
    )
    send_telegram(msg)

# For a 2GB VPS, use Qwen2.5-7B (INT4)
ollama pull qwen2.5:7b

# Verify
curl -s http://localhost:11434/api/tags | python3 -c "
import sys,json; d=json.load(sys.stdin)
print([m['name'] for m in d['models']])
"

# llm_analyzer.py

import json
import re
import requests
import logging

OLLAMA_URL = "http://localhost:11434/api/generate"
LOG = logging.getLogger(__name__)


def analyze_session_batch(
    session_id: str,
    commands: list[dict],
    baseline: dict,
    top_risk_events: list[dict],
) -> dict:
    cmds_text = "\n".join(
        f"[{c['ts']}] {c['cmd'][:80]}" for c in commands[:50]
    )

    prompt = f"""You are a Linux security audit expert.
Below is the last hour of SSH commands for VPS user ({baseline['user']}).

[User Behavioral Baseline]
- Command entropy: {baseline['cmd_entropy']} (0=highly repetitive, 1=fully random)
- Avg daily file writes: {baseline['file_write_rate']}
- Known outbound targets: {baseline['net_target_count']}
- Active hours: {active_hours(baseline)}

[Hard-Rule Hits]
{json.dumps(top_risk_events, ensure_ascii=False, indent=2) if top_risk_events else '(none)'}

[Command Log]
{cmds_text}

Output JSON only:
{{
  "risk_score": <0-100 integer>,
  "risk_level": "<low|medium|high|critical>",
  "reasoning": "<2-3 sentences, explaining the score>",
  "anomalies": [
    {{"cmd": "<specific command>", "issue": "<one-line explanation>"}}
  ],
  "suggestion": "<1-2 sentences for the ops engineer>"
}}"""

    try:
        resp = requests.post(
            OLLAMA_URL,
            json={
                "model": "qwen2.5:7b",
                "prompt": prompt,
                "stream": False,
                "options": {"temperature": 0.1, "num_predict": 1024},
            },
            timeout=120,
        )
        raw = resp.json().get("response", "")
        json_match = re.search(r"\{[\s\S]*\}", raw)
        return json.loads(json_match.group(0)) if json_match else {}
    except Exception as e:
        LOG.error(f"LLM analysis failed: {e}")
        return {"risk_score": 50, "risk_level": "unknown", "reasoning": str(e)}


def active_hours(baseline: dict) -> str:
    bins = baseline.get("hour_distribution", [0] * 24)
    active = [h for h, v in enumerate(bins) if v > 0]
    if not active:
        return "unknown"
    ranges = []
    start = active[0]
    for h in active[1:]:
        if h != start + 1:
            ranges.append((start, h - 1))
            start = h
    ranges.append((start, active[-1]))
    return ", ".join(f"{a:02d}:00-{b:02d}:59" for a, b in ranges)

# scheduler.py (pseudocode)

def scheduler_loop():
    while True:
        # ① Critical sessions first
        critical = get_unanalyzed_critical_sessions()
        for s in critical:
            result = analyze_session_batch(
                s["session_id"], s["commands"], s["baseline"], s["risk_events"]
            )
            save_llm_result(s["session_id"], result)
            if result.get("risk_level") in ("high", "critical"):
                send_telegram(format_llm_alert(s, result))

        # ② Regular batch (max 10 sessions per 5 min)
        batch = get_unanalyzed_sessions(limit=10)
        for s in batch:
            result = analyze_session_batch(
                s["session_id"], s["commands"], s["baseline"], []
            )
            save_llm_result(s["session_id"], result)

        time.sleep(300)

def generate_daily_report(date_str: str) -> str:
    conn = get_db_connection_for(date_str)
    total_sessions = count_sessions(conn)
    total_cmds     = count_commands(conn)
    high_risk      = count_risk_events(conn, "HIGH")
    llm_avg_score  = avg_llm_risk_score(conn)
    top_users      = top_active_users(conn, limit=5)

    return f"""📋 VPS SSH Audit Digest — {date_str}

📊 Overview
  Sessions: {total_sessions}  |  Total commands: {total_cmds}
  High-risk events: {high_risk}  |  LLM avg risk score: {llm_avg_score}/100

👤 Top 5 Active Users
{format_user_list(top_users)}

⚠️ High-Risk Events
{format_high_risk_events(get_high_risk_events(conn, limit=10))}

🤖 LLM Findings (Top 3)
{format_llm_findings(get_top_llm_findings(conn, limit=3))}

———
Auto-generated by local LLM. No data left the server.
"""

# telegram_alert.py

import requests
import os
import logging

LOG = logging.getLogger(__name__)
BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
CHAT_ID   = os.environ["TELEGRAM_CHAT_ID"]
API_URL   = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"

def send_telegram(text: str, parse_mode: str = "HTML"):
    resp = requests.post(
        API_URL,
        json={
            "chat_id": CHAT_ID,
            "text": text,
            "parse_mode": parse_mode,
            "disable_web_page_preview": True,
        },
        timeout=15,
    )
    if resp.status_code != 200:
        LOG.error(f"Telegram send failed: {resp.status_code} {resp.text}")
    else:
        LOG.info("Telegram alert sent")

#!/bin/bash
set -e

echo "=== 1. Install dependencies ==="
apt-get update -qq
apt-get install -y -qq python3-pip sqlite3 auditd
pip3 install requests

echo "=== 2. Pull Ollama model ==="
if ! command -v ollama &>/dev/null; then
    curl -fsSL https://ollama.com/install.sh | sh
    systemctl enable --now ollama
fi
ollama pull qwen2.5:7b
echo "   Model ready"

echo "=== 3. Create directories ==="
mkdir -p /opt/ssh_audit/{data,scripts}
# Copy your scripts here
# cp collector.py baseline.py risk_rules.py llm_analyzer.py \
#    telegram_alert.py scheduler.py /opt/ssh_audit/scripts/

echo "=== 4. Configure auditd ==="
cat > /etc/audit/rules.d/vps_ssh_audit.rules << 'EOF'
-w /var/log/auth.log -p wa -k ssh_auth
-a always,exit -F arch=b64 -S execve -F exit=0 -k vps_cmd
-w /root/.ssh -w /etc -w /opt -w /srv -w /var/www -p wa -k vps_file
-a always,exit -F arch=b64 -S connect -F exit=0 -k vps_net
EOF

echo "=== 5. Enable systemd services ==="
cat > /etc/systemd/system/ssh-audit.service << 'EOF'
[Unit]
Description=SSH Session Audit Daemon
After=network.target auditd.service
Requires=auditd.service

[Service]
ExecStart=/usr/bin/python3 /opt/ssh_audit/scripts/collector.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF

cat > /etc/systemd/system/ssh-audit-scheduler.service << 'EOF'
[Unit]
Description=SSH Audit LLM Scheduler
After=ssh-audit.service ollama.service
Requires=ollama.service

[Service]
ExecStart=/usr/bin/python3 /opt/ssh_audit/scripts/scheduler.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

systemctl daemon-reload
systemctl enable --now ssh-audit.service
systemctl enable --now ssh-audit-scheduler.service

echo "=== 6. Configure Telegram env ==="
cat > /etc/ssh_audit_env << EOF
TELEGRAM_BOT_TOKEN=your_b...here
EOF
chmod 600 /etc/ssh_audit_env

echo "=== Deployment complete ==="
systemctl status ssh-audit.service --no-pager
echo "Verify: journalctl -u ssh-audit -f"