sentinel/http_honeypot.py at master · webwizardg99/sentinel · GitHub

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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
#!/usr/bin/env python3
"""
SENTINEL HTTP Honeypot
Fake admin login panels on port 8080 — captures credentials, fingerprints, paths
Logs to /home/wizardg/sentinel/logs/http_honeypot.json (NDJSON)
"""
import asyncio
import json
import os
import time
import hashlib
from datetime import datetime, timezone
from pathlib import Path
from aiohttp import web

LOG_FILE = Path("/home/wizardg/sentinel/logs/http_honeypot.json")
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)

FAKE_PAGES = {
    "/admin": "admin_panel",
    "/login": "generic_login",
    "/wp-admin": "wordpress",
    "/wp-login.php": "wordpress",
    "/phpmyadmin": "phpmyadmin",
    "/manager/html": "tomcat",
    "/console": "jboss",
    "/.env": "env_file",
    "/config.php": "config_leak",
    "/backup": "backup",
    "/shell": "webshell",
    "/cmd": "webshell",
}

ADMIN_HTML = """<!DOCTYPE html><html><head><title>Admin Panel</title>
<style>body{{background:#1a1a2e;color:#e0e0e0;font-family:sans-serif;display:flex;align-items:center;justify-content:center;height:100vh;margin:0}}
.box{{background:#16213e;padding:40px;border-radius:8px;width:320px;box-shadow:0 4px 24px rgba(0,0,0,.5)}}
h2{{margin:0 0 24px;text-align:center;color:#4fc3f7}}
input{{width:100%;padding:10px;margin:8px 0;background:#0f3460;border:1px solid #4fc3f7;color:#fff;border-radius:4px;box-sizing:border-box}}
button{{width:100%;padding:12px;background:#4fc3f7;color:#000;font-weight:bold;border:none;border-radius:4px;cursor:pointer;margin-top:8px}}
.err{{color:#ef5350;font-size:.85em;text-align:center;margin-top:8px}}
</style></head><body>
<div class="box"><h2>{title}</h2>
<form method="POST"><input name="username" placeholder="Username" autocomplete="off">
<input name="password" type="password" placeholder="Password">
<button type="submit">Login</button></form>
{error}
</div></body></html>"""

WP_HTML = """<!DOCTYPE html><html><head><title>WordPress &rsaquo; Log In</title>
<style>body{{background:#f1f1f1;font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,sans-serif}}
#login{{width:320px;margin:80px auto;padding:26px 24px;background:#fff;border-radius:4px;box-shadow:0 1px 3px rgba(0,0,0,.13)}}
h1 a{{display:block;text-align:center;margin-bottom:16px;color:#1d2327;text-decoration:none;font-size:1.4em}}
input[type=text],input[type=password]{{width:100%;padding:8px;box-sizing:border-box;border:1px solid #8c8f94;border-radius:3px;margin:4px 0 12px}}
input[type=submit]{{background:#2271b1;color:#fff;border:none;padding:10px 20px;border-radius:3px;cursor:pointer;width:100%}}
</style></head><body><div id="login"><h1 a href="#">WordPress</h1>
<form method="POST">
<label>Username or Email<input type="text" name="log"></label>
<label>Password<input type="password" name="pwd"></label>
<input type="submit" name="wp-submit" value="Log In">
</form></div></body></html>"""

ENV_BODY = """DB_HOST=localhost
DB_NAME=production_db
DB_USER=admin
DB_PASS=Sup3rS3cr3t!
APP_KEY=base64:kQjF2oX8nVz+pLm/wRt1YdJeCs3Ah7GxBqMiNuEf5PO=
JWT_SECRET=8f4e2a9c1d6b3e7f0a5c8e2d9f1b4a7c
REDIS_URL=redis://127.0.0.1:6379
AWS_KEY=AKIAIOSFODNN7EXAMPLE
AWS_SECRET=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
MAIL_PASSWORD=EmailPass123!
"""


def _log(event: dict):
    event["@timestamp"] = datetime.now(timezone.utc).isoformat()
    with open(LOG_FILE, "a") as f:
        f.write(json.dumps(event) + "\n")


def _fingerprint(request: web.Request) -> dict:
    return {
        "ip": request.headers.get("X-Forwarded-For", request.remote),
        "ua": request.headers.get("User-Agent", ""),
        "referer": request.headers.get("Referer", ""),
        "accept_lang": request.headers.get("Accept-Language", ""),
        "host_header": request.headers.get("Host", ""),
    }


async def handle_get(request: web.Request):
    path = request.path.rstrip("/") or "/"
    page_type = FAKE_PAGES.get(path, "unknown_probe")
    fp = _fingerprint(request)

    _log({
        "event_type": "http_probe",
        "method": "GET",
        "path": path,
        "page_type": page_type,
        **fp,
    })

    if path in ("/.env",):
        return web.Response(text=ENV_BODY, content_type="text/plain")

    if "wp" in path:
        return web.Response(text=WP_HTML, content_type="text/html")

    titles = {
        "phpmyadmin": "phpMyAdmin",
        "tomcat": "Tomcat Manager",
        "jboss": "JBoss Console",
        "backup": "Backup Manager",
    }
    title = titles.get(page_type, "Administration")
    return web.Response(
        text=ADMIN_HTML.format(title=title, error=""),
        content_type="text/html"
    )


async def handle_post(request: web.Request):
    path = request.path.rstrip("/") or "/"
    fp = _fingerprint(request)

    try:
        data = await request.post()
        # Try multiple field name conventions
        username = (
            data.get("username") or data.get("log") or
            data.get("user") or data.get("email") or ""
        )
        password = (
            data.get("password") or data.get("pwd") or
            data.get("pass") or data.get("passwd") or ""
        )
    except Exception:
        username = password = ""

    _log({
        "event_type": "credential_attempt",
        "method": "POST",
        "path": path,
        "username": username,
        "password": password,
        "pw_hash": hashlib.sha256(password.encode()).hexdigest()[:16] if password else "",
        **fp,
    })

    # Always return "wrong password" after a short delay
    await asyncio.sleep(1.2)
    if "wp" in path:
        error_html = '<div style="color:#cc0000;font-size:.9em;margin-top:8px">Error: Incorrect username or password.</div>'
        return web.Response(text=WP_HTML.replace("</form>", f"</form>{error_html}"), content_type="text/html")

    return web.Response(
        text=ADMIN_HTML.format(title="Administration", error='<div class="err">Invalid credentials. Please try again.</div>'),
        content_type="text/html"
    )


async def handle_any(request: web.Request):
    path = request.path
    fp = _fingerprint(request)
    _log({
        "event_type": "http_probe",
        "method": request.method,
        "path": path,
        "page_type": "unknown_probe",
        **fp,
    })
    # 404 that looks real
    return web.Response(status=404, text="Not Found")


def create_app():
    app = web.Application()
    for path in FAKE_PAGES:
        app.router.add_get(path, handle_get)
        app.router.add_post(path, handle_post)
    app.router.add_route("*", "/{path_info:.*}", handle_any)
    return app


if __name__ == "__main__":
    print("[SENTINEL] HTTP Honeypot — port 8080")
    web.run_app(create_app(), host="0.0.0.0", port=8080, access_log=None)