AI Honeypot

Όχι άλλος ένας άδειος server

Ένα κλασικό honeypot καταγράφει usernames, passwords και commands. Χρήσιμο, αλλά παθητικό. Ο attacker καταλαβαίνει γρήγορα ότι κάτι δεν κολλάει: τα ίδια responses, ελάχιστα αρχεία, μη ρεαλιστικά errors. Το AI honeypot προσθέτει μια ελεγχόμενη στρώση προσαρμογής. Δεν αφήνει το μοντέλο να εκτελέσει εντολές. Το χρησιμοποιεί για να επιλέγει ή να συνθέτει ακίνδυνες, πειστικές αποκρίσεις μέσα σε αυστηρό sandbox.

Το σύστημα που θα στήσουμε δέχεται SSH και HTTP traffic σε decoy υπηρεσίες, μετατρέπει κάθε interaction σε κοινό event schema, εξάγει features και υπολογίζει risk score. Ένα local LLM εμπλουτίζει την εμπειρία του decoy, ενώ Elasticsearch και Kibana αναλαμβάνουν αναζήτηση, συσχετίσεις και real-time οπτικοποίηση.

Το threat model

Θεωρούμε ότι ο αντίπαλος μπορεί να σαρώσει ports, να δοκιμάσει credentials, να στείλει web probes, να κάνει directory discovery ή να ανεβάσει payload. Δεν εμπιστευόμαστε κανένα input. Το honeypot δεν έχει secrets, production credentials ή διαδρομή προς εσωτερικά συστήματα. Αν παραβιαστεί πλήρως, το blast radius πρέπει να παραμένει σχεδόν μηδενικό.

Η παγίδα σε πέντε στρώσεις

Το σημαντικό design decision είναι ο διαχωρισμός capture, enrich, store. Αν πέσει το LLM, το honeypot συνεχίζει με deterministic responses. Αν καθυστερήσει το Elasticsearch, η ουρά απορροφά προσωρινά events. Καμία εξωτερική εξάρτηση δεν πρέπει να κρατά ανοιχτό το network handler.

Πρώτα το κλουβί, μετά το δόλωμα

Στήσε το project σε ξεχωριστή VM ή isolated host. Το inbound traffic μπορεί να φτάνει μόνο στα decoy ports. Το outbound traffic κόβεται εξ ορισμού και επιτρέπεται μόνο προς το εσωτερικό Elasticsearch endpoint, DNS resolver και package mirror κατά την εγκατάσταση. Χρησιμοποίησε non-root user, read-only filesystem όπου γίνεται και αυστηρά CPU, RAM και process limits.

• Ξεχωριστό VLAN ή cloud subnet. Καμία route προς production, VPN ή management δίκτυο.
• Egress deny by default. Ένα captured malware sample δεν πρέπει να μπορεί να κάνει callback.
• Disposable VM snapshots. Επαναφορά σε καθαρή κατάσταση αντί για χειροκίνητο cleanup.
• Χωρίς πραγματικά δεδομένα. Τα fake αρχεία δεν περιέχουν ονόματα πελατών, API keys ή αληθινά domains.
• Σύντομο retention. Κράτα raw IP και payload μόνο όσο τεκμηριωμένα χρειάζονται.

Ένα schema για κάθε επίθεση

Το κοινό event format κάνει το SSH και το HTTP συγκρίσιμα. Κράτα timestamps σε UTC, ένα τυχαίο session ID και ξεχώρισε τα raw δεδομένα από τα derived features. Μην αφήνεις το attacker input να γίνει index name, filename ή log format string.

from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Any
import uuid

@dataclass(slots=True)
class AttackEvent:
    service: str
    source_ip: str
    action: str
    payload: dict[str, Any]
    session_id: str
    risk_score: int = 0
    ts: str = ""

    def to_document(self) -> dict[str, Any]:
        doc = asdict(self)
        doc["@timestamp"] = self.ts or datetime.now(timezone.utc).isoformat()
        doc.pop("ts", None)
        return doc

def new_session_id() -> str:
    return uuid.uuid4().hex

Ο collector χρησιμοποιεί bounded queue. Αυτό είναι μικρή αλλά κρίσιμη λεπτομέρεια: ένας flooder δεν πρέπει να γεμίσει όλη τη RAM. Αν η ουρά γεμίσει, κάνε drop χαμηλής αξίας events, μέτρησέ τα και κράτα summary counter.

import asyncio
from collections import Counter

EVENTS: asyncio.Queue = asyncio.Queue(maxsize=5000)
DROPPED = Counter()

async def publish(event) -> None:
    try:
        EVENTS.put_nowait(event)
    except asyncio.QueueFull:
        DROPPED[event.service] += 1

async def worker(enrich, store) -> None:
    while True:
        event = await EVENTS.get()
        try:
            enriched = await asyncio.wait_for(enrich(event), timeout=1.5)
            await store(enriched)
        except Exception:
            # Στο production: structured internal error, χωρίς raw secrets.
            pass
        finally:
            EVENTS.task_done()

Ένα shell που δεν εκτελεί τίποτα

Το Paramiko μπορεί να υλοποιήσει SSH server interface, αλλά το transport είναι blocking. Ο καθαρός τρόπος είναι να κρατήσεις το asyncio ως orchestrator και να μεταφέρεις το blocking session handler σε περιορισμένο thread pool. Για production-grade deployment αξίζει να μελετήσεις ώριμα projects όπως Cowrie, όμως εδώ χτίζουμε τον ελάχιστο μηχανισμό για να καταλάβουμε τη ροή.

Η σημαντικότερη αρχή: ποτέ subprocess, ποτέ os.system, ποτέ πραγματικό shell. Κάθε command περνά από parser, allowlist και fake filesystem. Άγνωστα commands επιστρέφουν ασφαλές error ή μια σύντομη απάντηση από το local LLM.

import shlex

FILES = {
  "/etc/hostname": "edge-cache-03\n",
  "/etc/os-release": 'NAME="Ubuntu"\nVERSION="22.04.5 LTS"\n',
  "/var/log/app.log": "2026-08-02 worker ready\n"
}

def respond(raw: str) -> tuple[str, dict]:
    try:
        argv = shlex.split(raw[:512])
    except ValueError:
        return "bash: syntax error\n", {"kind": "parse_error"}
    if not argv:
        return "", {"kind": "empty"}

    cmd, *args = argv
    if cmd == "whoami": return "deploy\n", {"kind": "discovery"}
    if cmd == "pwd": return "/srv/app\n", {"kind": "discovery"}
    if cmd == "uname": return "Linux edge-cache-03 5.15.0 x86_64\n", {"kind": "discovery"}
    if cmd == "cat" and args:
        body = FILES.get(args[0])
        return (body or f"cat: {args[0]}: No such file\n", {"kind": "file_read"})
    return f"{cmd}: command not found\n", {"kind": "unknown"}

Για credentials, αποδέξου ελεγχόμενα όλες τις προσπάθειες ή ένα μικρό σύνολο συνηθισμένων decoy combinations, αλλά κατέγραψε password hashes μόνο αν η πολιτική σου το επιτρέπει. Συνήθως αρκούν length, character classes και ένα salted fingerprint για deduplication. Δεν χρειάζεσαι μια βάση γεμάτη κλεμμένα passwords.

Ένα API που λέει αρκετά, ποτέ πολλά

Ο HTTP decoy μιμείται μια παλιά admin εφαρμογή με login, health endpoint και API routes. Η πειστικότητα δεν έρχεται από πραγματικές ευπάθειες. Έρχεται από συνεπή headers, latency, status codes και state μέσα στην ίδια session. Μπορείς να χρησιμοποιήσεις το χαμηλού επιπέδου asyncio.start_server για εκπαίδευση, αλλά για σωστό HTTP parsing προτίμησε μια ασφαλή βιβλιοθήκη εφαρμογών πίσω από reverse proxy.

from aiohttp import web
import asyncio, secrets

routes = web.RouteTableDef()

@routes.get("/api/health")
async def health(request):
    await asyncio.sleep(0.04)
    return web.json_response({"status":"ok", "node":"edge-cache-03"})

@routes.post("/admin/login")
async def login(request):
    data = await request.post()
    await request.app["publish_http"](request, {
      "action": "login_attempt",
      "username": str(data.get("username", ""))[:80],
      "password_length": len(str(data.get("password", "")))
    })
    response = web.HTTPFound("/admin/dashboard")
    response.set_cookie("session", secrets.token_urlsafe(24), httponly=True, samesite="Strict")
    raise response

app = web.Application(client_max_size=64 * 1024)
app.add_routes(routes)

Βάλε όριο μεγέθους body, timeout, rate limit ανά πηγή και συνολικό concurrency cap. Μην αποθηκεύεις binary uploads μέσα στο Elasticsearch. Υπολόγισε hash, MIME guess και μέγεθος, μετά μετέφερε το sample σε κλειστό quarantine storage μόνο αν έχεις ασφαλή διαδικασία malware handling.

Το μοντέλο ως ηθοποιός, όχι ως sysadmin

Το local LLM δεν βλέπει host tools και δεν αποκτά network access. Λαμβάνει ένα μικρό, καθαρισμένο context: τύπο command, fake hostname, fake current directory και τα τελευταία τρία safe interactions. Η έξοδός του περνά από length limit, control-character filter και policy validator. Αν κάτι αποτύχει, το deterministic shell παίρνει τον έλεγχο.

import re

SAFE_OUTPUT = re.compile(r"^[\x20-\x7E\n\t]{1,600}$")
BLOCKED = ("http://", "https://", "BEGIN PRIVATE KEY", "curl ", "wget ")

def validate_llm_output(text: str, fallback: str) -> str:
    text = text.strip()[:600]
    if not SAFE_OUTPUT.fullmatch(text):
        return fallback
    lowered = text.lower()
    if any(token.lower() in lowered for token in BLOCKED):
        return fallback
    return text + "\n"

Μην καλείς το μοντέλο σε κάθε keystroke. Κάλεσέ το μόνο για άγνωστα αλλά ακίνδυνα commands και βάλε cache ανά normalized input. Μικρό quantized model 3B έως 8B αρκεί, επειδή η εργασία είναι constrained role-play, όχι βαθιά ανάλυση. Για classification, ένα απλό rule engine ή μικρό sklearn model είναι συχνά ταχύτερο και πιο προβλέψιμο.

Από commands σε πρόθεση

Το risk score δεν χρειάζεται να ξεκινήσει με νευρωνικό δίκτυο. Φτιάξε πρώτα διαφανή features και βάρη. Μέτρα ταχύτητα commands, αποτυχημένα logins, discovery breadth, attempts για persistence, encoded strings και αλλαγές πρωτοκόλλου από την ίδια πηγή. Αργότερα, όταν έχεις labels, δοκίμασε Isolation Forest για anomalies ή supervised classifier για γνωστές κατηγορίες.

WEIGHTS = {
  "discovery": 8, "auth_burst": 20, "encoded_payload": 25,
  "persistence": 30, "cross_service": 15
}

def risk_score(features: set[str]) -> int:
    raw = sum(WEIGHTS.get(feature, 0) for feature in features)
    return min(100, raw)

def severity(score: int) -> str:
    if score >= 70: return "critical"
    if score >= 40: return "high"
    if score >= 20: return "medium"
    return "low"

Μην παρουσιάσεις το score ως βεβαιότητα. Είναι εργαλείο triage. Ένα security researcher, scanner ή λάθος monitoring job μπορεί να μοιάζει επιθετικό. Τα αυτόματα blocks χρειάζονται ξεχωριστά thresholds, expiry και human review, ειδικά όταν επηρεάζουν κοινόχρηστα VPN ή cloud ranges.

Το dashboard πρέπει να απαντά, όχι να διακοσμεί

Στο Elasticsearch χρησιμοποίησε index template με σωστά types. Τα IPs πρέπει να είναι ip, τα timestamps date, τα categories keyword και τα μεγάλα raw payloads μη indexed. Ενεργοποίησε Index Lifecycle Management ώστε τα raw events να διαγράφονται ή να συμπιέζονται αυτόματα.

{
  "index_patterns": ["honeypot-events-*"],
  "template": { "mappings": { "dynamic": "strict", "properties": {
    "@timestamp": {"type": "date"},
    "source_ip": {"type": "ip"},
    "service": {"type": "keyword"},
    "action": {"type": "keyword"},
    "session_id": {"type": "keyword"},
    "risk_score": {"type": "integer"},
    "payload.command": {"type": "wildcard"},
    "payload.raw": {"type": "object", "enabled": false}
  }}}
}

Τα έξι panels που αξίζουν

1. Sessions ανά service και ώρα: ξεχωρίζει scans από επίμονη δραστηριότητα.
2. Risk distribution: δείχνει αν τα thresholds είναι χρήσιμα ή όλα βαφτίζονται critical.
3. Top normalized commands: αφαιρεί μεταβλητά tokens για καθαρές τάσεις.
4. Session timeline: πλήρης ακολουθία ενεργειών για investigation.
5. Cross-service correlation: ίδια source IP σε SSH και HTTP μέσα σε δέκα λεπτά.
6. Collector health: queue depth, dropped events, LLM latency και indexing errors.

Το τελευταίο panel είναι το πιο παραμελημένο. Αν δεν ξέρεις ότι το pipeline χάνει events, το όμορφο threat map σου λέει ψέματα. Πρόσθεσε alert όταν η ουρά περάσει το 70%, όταν το indexing error rate αυξηθεί ή όταν ο decoy host ξεκινήσει outbound σύνδεση.

Δοκιμή χωρίς να παίξεις τον hacker

Ξεκίνα local και χρησιμοποίησε γνωστά, ακίνδυνα requests. Έλεγξε ότι οι υπηρεσίες απαντούν, τα events έχουν σωστό schema και το LLM fallback λειτουργεί όταν το model endpoint είναι offline. Μετά τρέξε load test μόνο από ελεγχόμενο host και με χαμηλό rate.

# Μόνο στο δικό σου isolated lab.
curl --max-time 2 http://127.0.0.1:8080/api/health
curl --max-time 2 -X POST http://127.0.0.1:8080/admin/login \
  -d 'username=demo&password=not-a-real-secret'

# Έλεγχος ότι δεν υπάρχει μη αναμενόμενο outbound traffic.
sudo ss -tpn state established

Γράψε unit tests για parser edge cases, oversized inputs, invalid UTF-8, prompt injection μέσα σε command strings και Elasticsearch downtime. Το honeypot πρέπει να αποτυγχάνει κλειστά: σύντομο generic response, ασφαλές log και κλείσιμο session. Όχι stack trace προς τον client.

Πριν ανοίξεις την πόρτα

• Τρέξε κάθε service με ξεχωριστό unprivileged user και dedicated directory.
• Βάλε reverse proxy ή firewall port redirect αντί να δώσεις privileged port στο Python process.
• Κράτα dependencies pinned και σκάναρε το image πριν από κάθε deployment.
• Εφάρμοσε rate limits, connection caps, idle timeout και συνολικό session timeout.
• Αφαίρεσε ANSI escapes και control characters από logs και LLM output.
• Μην εμπιστεύεσαι GeoIP, ASN ή reputation ως απόδειξη ταυτότητας.
• Κρυπτογράφησε τη μεταφορά προς Elasticsearch και χρησιμοποίησε index-only credentials.
• Όρισε retention, access logging και διαδικασία διαγραφής δεδομένων.
• Στήσε kill switch που κλείνει inbound και κάνει snapshot τα logs.

Πού μπαίνει πραγματικά το Machine Learning

Μετά από μερικές εβδομάδες καθαρών δεδομένων, ομαδοποίησε sessions με features όπως command n-grams, inter-arrival times, protocol sequence και error ratio. Ένα Isolation Forest μπορεί να εντοπίζει sessions που αποκλίνουν από τα συνηθισμένα bots. Ένα μικρό classifier μπορεί να κατηγοριοποιεί reconnaissance, credential attack, persistence probing και payload delivery. Κράτα explainable features και σύγκρινε πάντα με rule baseline.

Η σωστή φιλοδοξία δεν είναι «AI που πιάνει hackers». Είναι ένα παρατηρήσιμο, περιορισμένο σύστημα που αγοράζει χρόνο και παράγει καλύτερα σήματα. Το honeypot είναι αισθητήρας, όχι ασπίδα. Συνδύασέ το με patching, MFA, segmentation, backups και κανονικό incident response.