-
Notifications
You must be signed in to change notification settings - Fork 48
/
Copy pathhontel.py
149 lines (122 loc) · 4.58 KB
/
hontel.py
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
#!/usr/bin/env python
# Copyright (c) 2015 Miroslav Stampar (@stamparm)
# See the file 'LICENSE' for copying permission
import fcntl
import os
import re
import SocketServer
import stat
import subprocess
import threading
import time
try:
from telnetsrv.threaded import TelnetHandler, command
except ImportError:
exit("[!] please install telnetsrv (e.g. 'pip install telnetsrv')")
AUTH_USERNAME = None
AUTH_PASSWORD = None
WELCOME = None
LOG_PATH = "/var/log/%s.log" % os.path.split(__file__)[-1].split('.')[0]
READ_SIZE = 1024
CHECK_CHROOT = True
THREAD_DATA = threading.local()
LOG_FILE_PERMISSIONS = stat.S_IREAD | stat.S_IWRITE | stat.S_IRGRP | stat.S_IROTH
LOG_HANDLE_FLAGS = os.O_APPEND | os.O_CREAT | os.O_WRONLY
TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
USE_BUSYBOX = True
if CHECK_CHROOT:
chrooted = False
try:
output = subprocess.check_output("ls -di /", shell=True)
if int(output.split()[0]) != 2:
chrooted = True
except:
pass
finally:
if not chrooted:
exit("[!] run inside the chroot environment")
if USE_BUSYBOX:
try:
SHELL = "/bin/busybox sh"
_ = subprocess.check_output("/bin/busybox")
_ = _.split("\n")[0]
match = re.search(r".+\)", _)
if match:
_ = "%s built-in shell (ash)" % match.group(0)
WELCOME = "\n%s\nEnter 'help' for a list of built-in commands.\n" % _
except OSError:
exit("[!] please install busybox (e.g. 'apt-get install busybox')")
else:
SHELL = "/bin/bash"
class HoneyTelnetHandler(TelnetHandler):
WELCOME = WELCOME
PROMPT = "# "
authNeedUser = AUTH_USERNAME is not None
authNeedPass = AUTH_PASSWORD is not None
def _log(self, logtype, msg=None):
line = '[%s] [%s:%s] %s%s\n' % (time.strftime(TIME_FORMAT, time.localtime(time.time())), self.client_address[0], self.client_address[1], logtype, ": %s" % msg if msg is not None else "")
os.write(self._getLogHandle(), line)
def _getLogHandle(self):
if LOG_PATH != getattr(THREAD_DATA, "logPath", None):
if not os.path.exists(LOG_PATH):
open(LOG_PATH, "w+").close()
os.chmod(LOG_PATH, LOG_FILE_PERMISSIONS)
THREAD_DATA.logPath = LOG_PATH
THREAD_DATA.logHandle = os.open(THREAD_DATA.logPath, LOG_HANDLE_FLAGS)
return THREAD_DATA.logHandle
def _processRead(self):
result = ""
while self.process.poll() is None:
try:
buf = os.read(self.process.stdout.fileno(), READ_SIZE)
buf = re.sub(r"%s: line \d+: " % SHELL, "", buf)
result += buf
except OSError:
break
return result
def handleException(self, exc_type, exc_param, exc_tb):
return False
def session_start(self):
self._log("SESSION_START")
self.process = subprocess.Popen(SHELL, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
flags = fcntl.fcntl(self.process.stdout, fcntl.F_GETFL)
fcntl.fcntl(self.process.stdout, fcntl.F_SETFL, flags | os.O_NONBLOCK)
def session_end(self):
self._log("SESSION_END")
def handle(self):
if not self.authentication_ok():
return
if self.DOECHO and self.WELCOME:
self.writeline(self.WELCOME)
self.session_start()
while self.RUNSHELL and self.process.poll() is None:
line = self.input_reader(self, self.readline(prompt=self.PROMPT).strip())
raw = line.raw
cmd = line.cmd
params = line.params
self._log("CMD", raw)
if self.COMMANDS.has_key(cmd):
try:
self.COMMANDS[cmd](params)
continue
except:
pass
try:
self.process.stdin.write(raw.strip() + "\n")
except IOError, ex:
raise
finally:
time.sleep(0.1)
self.write(self._processRead())
def authCallback(self, username, password):
if username is not None and password is not None:
self._log("AUTH", "%s:%s" % (username, password))
if not(username == AUTH_USERNAME and password == AUTH_PASSWORD):
raise Exception("[x] wrong credentials ('%s':'%s')" % (username, password))
class TelnetServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
allow_reuse_address = True
server = TelnetServer(("0.0.0.0", 23), HoneyTelnetHandler)
try:
server.serve_forever()
except KeyboardInterrupt:
os._exit(1)